[luoluoyuyu]

Fix LOAD splitting of sparse batched aligned chunks (completes #19032)

Follow-up to #19032 by @Caideyipi. Commit 1 is the original fix, content unchanged,
authorship preserved. #19032's head branch lives inside apache/iotdb, so the follow-up work
cannot be pushed to it from a fork; this PR carries the whole change instead.
If the combined change is acceptable, #19032 can be closed.

TL;DR

BatchedAlignedValueChunkData rebuilds the standalone value chunks of a chunk group when LOAD
splits a TsFile by time partition. The old implementation assumed compact data and compact page
slots. For sparse data — which is exactly the layout the compaction writer produces for aligned
tables — it dropped null rows, dropped all-null pages, and could emit a page stream the reader
cannot pair with its time chunk. The visible result is wrong values, lost rows, or a split TsFile
that is silently truncated / cannot be parsed back.

The fix keeps three invariants while re-encoding:

1. one bitmap position per row — a null value still occupies one bit;
2. one page slot per time page — an all-null page still occupies its slot;
3. one byte per empty page — the empty page is the single 0 varint the format defines.

Contents

1. Background — when LOAD re-encodes a chunk, the compaction layout, the reader's three rules
2. Problems in master — P1 bitmap collapse, P2 lost page slots, P3 chunk-level decision, P4 bounds, P5 statistics
3. The fix — per-row writes, per-page sealing, one-byte empty pages, per-chunk layout choice
4. Tests
5. Validation
6. Scope and risk

---

1. Background

1.1 When LOAD re-encodes a chunk

LoadSingleTsFileNode#needDecodeTsFile decides that a TsFile must be split when a device spans
more than one time partition slot. TsFileSplitter then decodes the file and writes one new
TsFile per partition. For a chunk that lies inside a single partition the raw bytes are copied
(writeEntirePage); for a chunk (or page) that crosses a boundary the points are decoded and
re-encoded (writeDecodePage / writeDecodeValuePage).

1.2 The layout the compaction writer produces

To keep compaction memory bounded, the writer emits the first series batch together with the time
chunk and appends the remaining value chunks after all time chunks of the chunk group. That is
the "batched aligned" layout, and those trailing chunks are the ones this class rebuilds:

    aligned device (table "t", tag "tagA") — one chunk group
    ┌────────────────────────────────────────────────────────────────────────────┐
    │ TIME chunk (VECTOR)   │ T0 │ T1 │ T2 │ T3 │                                │
    ├────────────────────────────────────────────────────────────────────────────┤
    │ f1  VALUE chunk       │ V0 │ V1 │ V2 │ V3 │   ← old layout: value chunk    │
    │                       │    │    │    │    │     directly follows its time  │
    ├────────────────────────────────────────────────────────────────────────────┤
    │ f2  VALUE chunk       │ W0 │ W1 │ W2 │ W3 │   ← standalone: written after  │
    │     (no time chunk)   │    │    │    │    │     all time chunks            │
    ├────────────────────────────────────────────────────────────────────────────┤
    │ f3  VALUE chunk       │ X0 │ X1 │ X2 │ X3 │                                │
    └────────────────────────────────────────────────────────────────────────────┘

AlignedChunkData handles the f1 style with AlignedChunkWriterImpl (time and value chunks
interleaved). f2/f3 have no following time chunk, so the splitter cannot use that writer and
BatchedAlignedValueChunkData drives a plain ValueChunkWriter instead. Both paths must produce
chunks that the aligned reader accepts.

1.3 Why re-encoding is delicate: the reader's three rules

Rule A — the value page bitmap has one position per timestamp, nulls included

A value page body is:

    value page body
    ┌──────────┬────────────────────┬─────────────────────────────┐
    │ int size │ bitmap (size+7)/8  │ non-null values, in order   │
    └──────────┴────────────────────┴─────────────────────────────┘

    rows (timestamps)   t0    t1    t2    t3
    values               1   null    3   null
    bitmap              1     0     1     0      size = 4
    value stream        [1]               [3]

    reader: bit i  ↔  i-th timestamp of the paired time page

ValuePageWriter#setBit(isNull) only raises the bit for a non-null value, but size is
incremented for every row. The reader (ValuePageReader#splitDataToBitmapAndValue) indexes the
bitmap with timeIndex, i.e. with the position inside the time page.

Rule B — page k of a value chunk pairs with page k of its time chunk

AbstractAlignedChunkReader#initAllPageReaders drives the loop from the time chunk and takes
exactly one page from every value chunk per iteration:

    while (timeChunkDataBuffer.remaining() > 0) {
      PageHeader timePage = PageHeader.deserializeFrom(timeChunkDataBuffer, ...);
      for (Chunk value : valueChunks) {
        PageHeader valuePage = PageHeader.deserializeFrom(value.getData(), ...);
      }
    }

        T0        T1        T2        T3        <- time pages (drive the loop)
        │         │         │         │
        ▼         ▼         ▼         ▼
        V0        V1        V2        V3        <- value pages, paired by POSITION

The page count also changes how the chunk is parsed: ChunkHeader sets
numOfPages <= 1 ? ONLY_ONE_PAGE_CHUNK_HEADER : CHUNK_HEADER, and the reader has a separate
branch for single-page chunks. Declared page count and real page count must agree exactly.

Rule C — an all-null page is exactly one byte

The format's "this page is all null" page is a zero-length body whose header is the single varint
uncompressedSize = 0. PageHeader#deserializeFrom returns immediately in that case, leaving the
remaining bytes for the next page:

    legal empty page                          placeholder header with statistics
    bytes:  00                                bytes:  00  00
            └─ uncompressedSize = 0                   │   └─ compressedSize = 0 (extra!)
               reader: "empty page", done            └─ "empty page" #1

                                              reader: 00 → empty page #1
                                                      00 → empty page #2   ← phantom

Any extra byte turns into a phantom page and shifts every following page slot.

---

2. Problems in master

All of the following are in BatchedAlignedValueChunkData#encodeAndBuildChunks before this PR.

P1 — null rows were not written, so the bitmap collapsed

// before
for (int j = 0; j < page.timeBatch.length; j++) {
  long time = page.timeBatch[j];
  if (time >= startTime && time <= endTime) {
    boolean isNull = (page.valueBatch == null || page.valueBatch.length == 0 || page.valueBatch[j] == null);
    if (!isNull) {                                        // ← null rows skipped
      hasBufferedData = true;
      pageHasValue = true;
      valueChunkValueWriter.write(valueChunkWriter, 0, page.valueBatch[j], false);  // ← dummy time 0
    }
  }
}

Consequence — with v(t) = t + 1, null at odd timestamps, one page in partition [0, I-1]:

    timestamps      t0    t1    t2    t3
    real values      1   null    3   null

    written        size=2, bitmap=11, values=[1, 3]
    read back      t0 = 1 ✅   t1 = 3 ❌   t2 = null ❌   t3 = null ❌ (row lost)

Every value after the first null is attached to a previous timestamp, and the tail of the page
becomes null. This is the "wrong values / missing rows" symptom.

P2 — all-null pages were not written, so every later page shifted

Two different code paths dropped a page slot:

// before: a copied empty page is dropped
if (!page.needDecode) {
  if (page.pageData.remaining() > 0) {          // ← 0-byte page: nothing written
    valueChunkWriter.writePageHeaderAndDataIntoBuff(page.pageData, page.pageHeader);
    hasBufferedData = true;
  }
  continue;
}

// before: a decoded page with rows but no non-null value is not sealed
...
if (pageHasValue) {                             // ← false for an all-null page
  valueChunkWriter.sealCurrentPage();
}

Not sealing is worse than "dropping a page": ValuePageWriter is not reset either, so the next
source page's rows are appended to the same page writer and the page boundaries are lost as well.

Consequence — partition [0, I-1] with source pages {0,1} (all null) and {I-2, I-1} (values):

    time chunk   [ T0: rows 0, 1 ][ T1: rows I-2, I-1 ]
                        │                     │
                        ▼                     ▼
    value chunk  [ V0: values of I-2, I-1 ]   ✗  (nothing left)

                  one page instead of two: V0 is consumed as the counterpart of T0,
                  so rows 0/1 receive the values of I-2/I-1 and T1 reads nothing

P3 — the chunk-level decision conflated "wrote data" with "has rows"

hasBufferedData was used to answer two different questions: did the writer buffer anything
and does this chunk cover any row of this partition. For a partition whose pages are
[data][all null][data] the first question is true, so the chunk is written with pages — but the
all-null page was skipped (P2), producing 2 pages against 3 time pages:

    time chunk   [ T0 ][ T1 ][ T2 ]
    value chunk  [ V0 ][ ✗  ][ V2 ]        ← T2's values are paired with T1
    fixed        [ V0 ][ ∅  ][ V2 ]        ∅ = one 0x00 byte (all-null page)

Conversely, a chunk that has no row at all in the partition must keep the zero-page layout of
an all-null column — and must still be emitted: the measurements of a chunk group are aligned by
chunk position, so dropping the chunk entirely shifts every later measurement.

P4 — valueBatch may be shorter than timeBatch

boolean isNull = (page.valueBatch == null
    || page.valueBatch.length == 0      // ← only guards the empty array
    || page.valueBatch[j] == null);     // ← AIOOBE when j >= length

P5 — statistics were patched by hand around a synthetic timestamp

The dummy time 0 was compensated afterwards:

Statistics<? extends Serializable> statistics = valueChunkWriter.getPageWriter().getStatistics();
if (pageStartTime <= pageEndTime) {
  statistics.setStartTime(pageStartTime);
  statistics.setEndTime(pageEndTime);
}

This mutates the writer's internal page statistics, which the writer itself manipulates for its
"first page statistics are stored only at chunk level" optimization, and it is executed only for
pages that had at least one non-null value — i.e. exactly the pages that P1/P2 were about to
handle incorrectly.

---

3. The fix

3.1 Write every row, with its real timestamp (fixes P1, P5)

// after
boolean isNull = (page.valueBatch == null
    || j >= page.valueBatch.length      // P4
    || page.valueBatch[j] == null);
// The bitmap has one position per timestamp, including nulls. Use the real timestamp
// so the writer also computes the statistics of the non-null values correctly.
valueChunkValueWriter.write(valueChunkWriter, time, isNull ? null : page.valueBatch[j], isNull);
pageHasRows = true;

ValuePageWriter#write calls setBit(isNull) for every row (so the bitmap keeps one position per
timestamp) and statistics.update(time, value) only for non-null values (so page and chunk
statistics are computed by the writer, exactly as in the normal write path). The manual
setStartTime / setEndTime block disappears.

Also note the writer already understands all-null pages: sealCurrentPage seals when
pageWriter.getSize() != 0 (positions, nulls included), and
ValuePageWriter#writePageHeaderAndDataIntoBuff emits writeEmptyPageIntoBuff — one 0 byte —
when statistics.getCount() == 0. The format supported all-null pages all along; the splitter was
simply not producing them.

3.2 Seal every page that has rows (fixes P2)

// after
if (pageHasRows) {          // rows of this partition, nulls included
  valueChunkWriter.sealCurrentPage();
}

For a copied page, an empty page no longer disappears; it is re-emitted and keeps its slot:

// after
if (page.pageData.remaining() == 0) {
  if (chunkHasRows) {
    valueChunkWriter.writePageHeaderAndDataIntoBuff(EMPTY_PAGE_DATA, EMPTY_PAGE_HEADER);
  }
  continue;
}

EMPTY_PAGE_HEADER is new PageHeader(0, 0, null). The header must not carry statistics.
This is easy to get wrong: TsFileSplitter#handleEmptyValueChunk builds the placeholder page as
new PageHeader(0, 0, chunkMetadata.getStatistics()), and for the first page of a chunk
ValueChunkWriter#writePageHeaderAndDataIntoBuff(data, header) writes a compressedSize varint in
addition to uncompressedSize whenever header.getStatistics() != null, deferring those
statistics into firstPageStatistics to splice them into the buffer later. Re-emitting the
placeholder header as the first page of a chunk therefore produces two bytes where the format
defines one, and the reader parses a phantom second empty page — the real data page then has no
slot and the column silently reads as all null. That is why the fix normalizes the header instead
of reusing it.

3.3 Decide per chunk: mirror the page slots or keep the zero-page layout (fixes P3)

The coarse hasBufferedData flag is replaced by a predicate that answers the actual question —
does any page of this chunk hold rows of this partition?

private static boolean hasPageWithRows(final List<PageBuffer> chunkPages) {
  for (final PageBuffer page : chunkPages) {
    // A decoded page already records how many of its rows belong to this partition, and a page
    // copied as it is was read from this partition, so a body means rows and only an all-null
    // page has no body.
    if (page.needDecode ? page.satisfiedLength > 0 : page.pageData.remaining() > 0) {
      return true;
    }
  }
  return false;
}

Both page kinds answer the question from data the splitter has already recorded, so no timestamp is
visited a second time: PageBuffer.satisfiedLength is the number of rows the page contributes to
the partition, filled in by TsFileSplitter while it splits the page
(writeDecodePage(times, values, satisfiedLength)), and a copied page was read from this partition
as a whole, so its body is non-empty exactly when it holds rows. This is the same test the base
class uses in AlignedChunkData#hasPageWithData.

The result selects between the two layouts the reader understands:

    chunkHasRows = true  → mirror the time chunk's page slots
        time   │ T0 │ T1 │ T2 │
        value  │ V0 │ ∅  │ V2 │        ∅ = 0x00, "this page is all null"
        dataSize = getByteBuffer(), pageCount = getNumOfPages()

    chunkHasRows = false → zero-page chunk (all-null column layout), chunk is still emitted
        time   │ T0 │ T1 │
        value  (no pages)              metadata count = 0 → reader supplies nulls
        dataSize = 0, pageCount = 0

This is the counterpart of the rule the base class already applies in
AlignedChunkData#buildChunkFromUndecodedPages (which returns null and falls back to the writer
path when a chunk consists only of empty pages). Before this change, the fast path obeyed the
rules and the writer path — the one taken as soon as a single page must be decoded — did not.

3.4 The resulting loop

final boolean chunkHasRows = hasPageWithRows(chunkPages, startTime, endTime);
for (final PageBuffer page : chunkPages) {
  if (!page.needDecode) {
    if (page.pageData.remaining() == 0) {
      if (chunkHasRows) {
        valueChunkWriter.writePageHeaderAndDataIntoBuff(EMPTY_PAGE_DATA, EMPTY_PAGE_HEADER);
      }
      continue;
    }
    valueChunkWriter.writePageHeaderAndDataIntoBuff(page.pageData, page.pageHeader);
    continue;
  }

  boolean pageHasRows = false;
  for (int j = 0; j < page.timeBatch.length; j++) {
    final long time = page.timeBatch[j];
    if (time >= startTime && time <= endTime) {
      final boolean isNull = page.valueBatch == null
          || j >= page.valueBatch.length
          || page.valueBatch[j] == null;
      valueChunkValueWriter.write(
          valueChunkWriter, time, isNull ? null : page.valueBatch[j], isNull);
      pageHasRows = true;
    }
  }
  if (pageHasRows) {
    valueChunkWriter.sealCurrentPage();
  }
}

final ByteBuffer valueData =
    chunkHasRows ? valueChunkWriter.getByteBuffer() : EMPTY_CHUNK_DATA;
final ChunkHeader encodedHeader =
    createEncodedChunkHeader(
        chunkHeader,
        valueData.remaining(),
        chunkHasRows ? valueChunkWriter.getNumOfPages() : 0,
        VALUE_COLUMN_MASK);

Change map

Location	Change	Problem
BatchedAlignedValueChunkData constants	add EMPTY_PAGE_DATA, EMPTY_PAGE_HEADER	P2 (byte-level)
encodeAndBuildChunks()	chunkHasRows replaces hasBufferedData	P3
copied-page branch	keep empty page, normalize its header	P2
decoded-page branch	write every row, real timestamp, pageHasRows, bounds guard	P1, P4, P5
decoded-page branch	remove pageStartTime / pageEndTime / statistics patch	P5
chunk assembly	dataSize / pageCount from chunkHasRows	P3
new helper	hasPageWithRows(...)	P3

No format change, no change to the normal write/compaction path, no config, no public API.

---

4. Tests

4.1 TsFileSplitterTest — five new cases, one shared harness

assertSparseBatchedAlignedChunkSplit(pages, isNull) builds the compaction layout with
TsFileIOWriter (time chunk + f1 through AlignedChunkWriterImpl, then f2 through a
standalone ValueChunkWriter), verifies the source file reads back as built, runs the real
splitter, then for every partition writes the target TsFile and reads it back row by row.

Besides the row assertions, the harness checks that every chunk declares exactly the bytes it
carries:

Assert.assertEquals(
    "chunk of measurement '" + chunk.getHeader().getMeasurementID()
        + "' declares a data size that does not match the bytes it carries",
    chunk.getHeader().getDataSize(),
    chunk.getData().remaining());

Test	Page shape / null pattern	Covers
...PreservesNullRows	one page, 7 rows, nulls on odd timestamps, spans 2 partitions	P1 — bitmap positions
...PreservesCopiedEmptyPage	{0,1} all null (copied) + a page crossing the boundary	P2 — copied empty page slot
...PreservesDecodedEmptyPage	a decoded page that is all null, crossing 2 boundaries	P2 — decoded all-null page
...PreservesCopiedPageBeforeDecodedPage	copied page followed by a decoded page in one partition	statistics splicing, exact page count
...PreservesEmptyPageAfterDataPages	[data][all null][data]	P3 — chunk-level decision

4.2 IoTDBLoadTsFileIT#testLoadSparseBatchedAlignedTsFile

Builds the same layout as a real file whose three pages span three configured partitions — one
page inside a partition (copied), one crossing a boundary with a null row (decoded), one entirely
null (empty page slot) — then goes through SQL only:

load '<file>' with ('database'='root');
select time, f1, f2 from test order by time;

Every row's timestamp, f1 and f2 is asserted, including resultSet.wasNull() for the null
rows. This is the only test that covers LOAD's scheduling, the partition configuration and the
metadata index of the produced files.

---

5. Validation

• mvn spotless:check -pl iotdb-core/datanode (and the same for integration-test with
  -P with-integration-tests): BUILD SUCCESS.
• mvn validate -pl integration-test -P with-integration-tests: 0 Checkstyle violations.
• The IT fixture was verified outside the cluster: writing it and reading it back with
  TsFileReader returns the expected rows with the expected nulls (this is how a missing
  endFile() in the fixture was caught).
• Not executed locally: this environment has pre-existing compilation errors in unrelated datanode
  test sources (stale ~/.m2 artifacts) and no thrift compiler for -am builds, so neither
  TsFileSplitterTest nor the IT can be started here; the changed files were type-checked with
  javac against the module classpath instead.
• CI: the first run of this PR failed during testCompile of iotdb-server on
  WALNodeTest.java:366 (cannot find symbol: walFlushListeners), which is a defect introduced on
  master by Fix flaky WALNode outdated file deletion test #17921 and fixed there by Fix WALNodeTest compilation after explicit WAL rolling #19189. That failure is unrelated to this PR and happened
  before any test of this change could run; the branch has been updated with master.

---

6. Scope and risk

• The only production change is the encodeAndBuildChunks() override of one class, used
  exclusively by TsFileSplitter when LOAD has to re-encode a standalone value chunk.
• No change to the file format, the normal write path, compaction, the base AlignedChunkData
  behaviour, configuration or public API.
• The change makes the splitter stricter rather than more permissive: chunks now declare the exact
  page count and byte count they contain, and every page slot of the time chunk has a counterpart.
  A file that used to be produced with shifted pages now fails loudly (header/data mismatch) or
  reads correctly, instead of returning wrong values.

---

This PR:

• has been self-reviewed.
• adds comments explaining the "why" and the intent of the code where it is not obvious for an
  unfamiliar reader.
• adds unit tests covering the new code paths.
• adds integration tests.

Key changed/added classes

• org.apache.iotdb.db.storageengine.load.splitter.BatchedAlignedValueChunkData
• org.apache.iotdb.db.storageengine.load.TsFileSplitterTest
• org.apache.iotdb.relational.it.db.it.IoTDBLoadTsFileIT