Fix LongArraySource#fillPrevChunk When a Converter is Supplied (#4431)

deephaven · Sep 6, 2023 · 51f82f1 · 51f82f1
1 parent 564f824
commit 51f82f1
Show file tree

Hide file tree

Showing 3 changed files with 68 additions and 41 deletions.
diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/sources/LongArraySource.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/sources/LongArraySource.java
@@ -596,31 +596,20 @@ public <R> void fillPrevChunk(
  final long[] inUse = prevInUse[blockNo];
  if (inUse != null) {
  // region conditionalCopy
+ final int chunkOffset = destOffset.intValue();
  long[] baseInput = (long[]) getBlock(blockNo);
  long[] overInput = (long[]) getPrevBlock(blockNo);
- effectiveContext.copyKernel.conditionalCopy(destination, baseInput, overInput,
- inUse, srcOffset, destOffset.intValue(), length);
 
- int bitsSet = 0;
- final int bitsetLen = (length + 63) >> 6;
- final int bitsetOffset = srcOffset >> 6;
- for (int i = 0; i < bitsetLen; ++i) {
- bitsSet += Long.bitCount(inUse[i + bitsetOffset]);
- }
- final int totalBits = bitsetLen << 6;
- final boolean flipBase = bitsSet > totalBits / 2;
+ final int srcEndOffset = srcOffset + length;
+ int nextBit = CopyKernel.Utils.nextSetBit(inUse, srcOffset, srcEndOffset, false);
 
- // mem-copy from baseline
  for (int ii = 0; ii < length; ++ii) {
- chunk.set(destOffset.intValue() + ii, converter.apply((flipBase ? overInput : baseInput)[srcOffset + ii]));
- }
-
- final int srcEndOffset = srcOffset + length;
- for (int ii = CopyKernel.Utils.nextSetBit(inUse, srcOffset, srcEndOffset, flipBase);
- ii < srcEndOffset;
- ii = CopyKernel.Utils.nextSetBit(inUse, ii + 1, srcEndOffset, flipBase)) {
- chunk.set(destOffset.intValue() + ii - srcOffset,
- converter.apply(flipBase ? baseInput[ii] : overInput[ii]));
+ if (ii != nextBit - srcOffset) {
+ chunk.set(ii + chunkOffset, converter.apply(baseInput[ii + srcOffset]));
+ } else {
+ nextBit = CopyKernel.Utils.nextSetBit(inUse, nextBit + 1, srcEndOffset, false);
+ chunk.set(ii + chunkOffset, converter.apply(overInput[ii + srcOffset]));
+ }
  }
  // endregion conditionalCopy
  } else {

diff --git a/...ne/table/src/test/java/io/deephaven/engine/table/impl/select/TestReinterpretedColumn.java b/...ne/table/src/test/java/io/deephaven/engine/table/impl/select/TestReinterpretedColumn.java
@@ -5,7 +5,9 @@
 
 import io.deephaven.chunk.LongChunk;
 import io.deephaven.chunk.ObjectChunk;
+import io.deephaven.chunk.WritableObjectChunk;
 import io.deephaven.chunk.attributes.Values;
+import io.deephaven.engine.context.ExecutionContext;
 import io.deephaven.engine.rowset.RowSet;
 import io.deephaven.engine.rowset.RowSetFactory;
 import io.deephaven.engine.table.ChunkSource;
@@ -14,6 +16,7 @@
 import io.deephaven.engine.table.TableDefinition;
 import io.deephaven.engine.table.WritableColumnSource;
 import io.deephaven.engine.table.impl.QueryTable;
+import io.deephaven.engine.table.impl.sources.ArrayBackedColumnSource;
 import io.deephaven.engine.table.impl.sources.InstantArraySource;
 import io.deephaven.engine.table.impl.sources.InstantSparseArraySource;
 import io.deephaven.engine.table.impl.sources.LongArraySource;
@@ -23,6 +26,7 @@
 import io.deephaven.engine.table.impl.sources.ZonedDateTimeArraySource;
 import io.deephaven.engine.table.impl.sources.ZonedDateTimeSparseArraySource;
 import io.deephaven.engine.table.impl.util.TableTimeConversions;
+import io.deephaven.engine.testutil.ControlledUpdateGraph;
 import io.deephaven.engine.testutil.testcase.RefreshingTableTestCase;
 import io.deephaven.time.DateTimeUtils;
 import org.apache.commons.lang3.mutable.MutableInt;
@@ -425,4 +429,49 @@ private LocalTime makeLocalTime(int hour, int ii, boolean sorted) {
 
  return LocalTime.of(hour + hourOff, minute, 0);
  }
+
+ @Test
+ public void testFillPrevOnInstantColumn() {
+ final InstantArraySource source = new InstantArraySource();
+ source.startTrackingPrevValues();
+
+ final ControlledUpdateGraph updateGraph = ExecutionContext.getContext().getUpdateGraph().cast();
+
+ try (final RowSet latticeRows = RowSetFactory.flat(1024);
+ final RowSet unchangedRows = RowSetFactory.flat(1024).shift(ArrayBackedColumnSource.BLOCK_SIZE)) {
+ source.ensureCapacity(unchangedRows.lastRowKey() + 1);
+ updateGraph.runWithinUnitTestCycle(() -> {
+ latticeRows.forAllRowKeys(row -> {
+ source.set(row, Instant.ofEpochMilli(row));
+ });
+
+ // create rows that won't change in a separate block for code coverage
+ unchangedRows.forAllRowKeys(row -> {
+ source.set(row, Instant.ofEpochMilli(row));
+ });
+ });
+
+ updateGraph.runWithinUnitTestCycle(() -> {
+ // change only some rows; for coverage
+ latticeRows.forAllRowKeys(row -> {
+ if (row % 2 == 0) {
+ source.set(row, Instant.ofEpochMilli(row + latticeRows.size()));
+ }
+ });
+
+ try (final RowSet allRows = latticeRows.union(unchangedRows);
+ final ChunkSource.FillContext context = source.makeFillContext(allRows.intSize());
+ final WritableObjectChunk<Instant, Values> chunk =
+ WritableObjectChunk.makeWritableChunk(allRows.intSize())) {
+ source.fillPrevChunk(context, chunk, allRows);
+ allRows.forAllRowKeys(row -> {
+ final long chunkOffset = row + (row < ArrayBackedColumnSource.BLOCK_SIZE
+ ? 0
+ : (latticeRows.size() - ArrayBackedColumnSource.BLOCK_SIZE));
+ assertEquals(Instant.ofEpochMilli(row), chunk.get((int) chunkOffset));
+ });
+ }
+ });
+ }
+ }
 }
diff --git a/replication/static/src/main/java/io/deephaven/replicators/ReplicateSourcesAndChunks.java b/replication/static/src/main/java/io/deephaven/replicators/ReplicateSourcesAndChunks.java
@@ -1097,31 +1097,20 @@ private static List<String> addLongToBoxedAdapter(
  " data[segment][destOffset + jj] = converter.applyAsLong(chunk.get(offset + jj));",
  " }"));
  permuted = replaceRegion(permuted, "conditionalCopy", Arrays.asList(
+ " final int chunkOffset = destOffset.intValue();",
  " long[] baseInput = (long[]) getBlock(blockNo);",
  " long[] overInput = (long[]) getPrevBlock(blockNo);",
- " effectiveContext.copyKernel.conditionalCopy(destination, baseInput, overInput,",
- " inUse, srcOffset, destOffset.intValue(), length);",
- "",
- " int bitsSet = 0;",
- " final int bitsetLen = (length + 63) >> 6;",
- " final int bitsetOffset = srcOffset >> 6;",
- " for (int i = 0; i < bitsetLen; ++i) {",
- " bitsSet += Long.bitCount(inUse[i + bitsetOffset]);",
- " }",
- " final int totalBits = bitsetLen << 6;",
- " final boolean flipBase = bitsSet > totalBits / 2;",
- "",
- " // mem-copy from baseline",
- " for (int ii = 0; ii < length; ++ii) {",
- " chunk.set(destOffset.intValue() + ii, converter.apply((flipBase ? overInput : baseInput)[srcOffset + ii]));",
- " }",
  "",
  " final int srcEndOffset = srcOffset + length;",
- " for (int ii = CopyKernel.Utils.nextSetBit(inUse, srcOffset, srcEndOffset, flipBase);",
- " ii < srcEndOffset;",
- " ii = CopyKernel.Utils.nextSetBit(inUse, ii + 1, srcEndOffset, flipBase)) {",
- " chunk.set(destOffset.intValue() + ii - srcOffset,",
- " converter.apply(flipBase ? baseInput[ii] : overInput[ii]));",
+ " int nextBit = CopyKernel.Utils.nextSetBit(inUse, srcOffset, srcEndOffset, false);",
+ "",
+ " for (int ii = 0; ii < length; ++ii) {",
+ " if (ii != nextBit - srcOffset) {",
+ " chunk.set(ii + chunkOffset, converter.apply(baseInput[ii + srcOffset]));",
+ " } else {",
+ " nextBit = CopyKernel.Utils.nextSetBit(inUse, nextBit + 1, srcEndOffset, false);",
+ " chunk.set(ii + chunkOffset, converter.apply(overInput[ii + srcOffset]));",
+ " }",
  " }"));
  permuted = applyFixup(permuted, "copyFromArray",
  "^(\\s+).+\\.copyFromArray\\(([^,]+), ([^,]+), ([^,]+), ([^)]+)\\).*", (m) -> Arrays.asList(