diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java new file mode 100644 index 0000000000000..375f560c9ee41 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java @@ -0,0 +1,142 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.it; + +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; +import org.apache.iotdb.isession.ISession; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; + +@RunWith(IoTDBTestRunner.class) +@Category({LocalStandaloneIT.class}) +public class IoTDBAlignedTVListLazyAllocationIT { + + private static final String DEVICE = "root.aligned_lazy_allocation.d1"; + private static final int DATANODE_MAX_HEAP_SIZE_IN_MB = 256; + private static final int INITIAL_ROW_COUNT = 40_000; + private static final int NEW_COLUMN_COUNT = 1_200; + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getDataNodeJVMConfig() + .setMaxHeapSize(DATANODE_MAX_HEAP_SIZE_IN_MB); + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setAutoCreateSchemaEnabled(true) + .setEnableMemControl(true) + .setPrimitiveArraySize(64) + .setMemtableSizeThreshold(512L * 1024 * 1024) + .setDatanodeMemoryProportion("6:1:1:1:1:1") + .setWriteMemoryProportion("100:1"); + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testAddingManyColumnsAfterManyRowsDoesNotExhaustWriteMemory() throws Exception { + int historicalBlockCount = (INITIAL_ROW_COUNT + ARRAY_SIZE - 1) / ARRAY_SIZE; + long eagerAllocationCost = + (long) historicalBlockCount + * NEW_COLUMN_COUNT + * AlignedTVList.valueListArrayMemCost(TSDataType.INT64); + long lazyAllocationCost = + (long) historicalBlockCount + * NEW_COLUMN_COUNT + * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray() + + (long) NEW_COLUMN_COUNT * AlignedTVList.primitiveArrayMemCost(TSDataType.INT64); + long dataNodeMaxHeapSize = DATANODE_MAX_HEAP_SIZE_IN_MB * 1024L * 1024L; + + Assert.assertTrue( + "The eager implementation must exceed the entire DataNode heap in this scenario", + eagerAllocationCost > dataNodeMaxHeapSize); + Assert.assertTrue( + "The lazy implementation must fit comfortably within the DataNode heap", + lazyAllocationCost < dataNodeMaxHeapSize / 2); + + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + insertInitialRows(session); + + List measurements = new ArrayList<>(NEW_COLUMN_COUNT); + List dataTypes = new ArrayList<>(NEW_COLUMN_COUNT); + List values = new ArrayList<>(NEW_COLUMN_COUNT); + for (int i = 1; i <= NEW_COLUMN_COUNT; i++) { + measurements.add("s" + i); + dataTypes.add(TSDataType.INT64); + values.add((long) i); + } + session.insertAlignedRecord(DEVICE, INITIAL_ROW_COUNT, measurements, dataTypes, values); + + try (SessionDataSet dataSet = + session.executeQueryStatement( + "SELECT COUNT(s0), COUNT(s" + NEW_COLUMN_COUNT + ") FROM " + DEVICE)) { + Assert.assertTrue(dataSet.hasNext()); + List fields = dataSet.next().getFields(); + Assert.assertEquals(INITIAL_ROW_COUNT, fields.get(0).getLongV()); + Assert.assertEquals(1, fields.get(1).getLongV()); + Assert.assertFalse(dataSet.hasNext()); + } + } + } + + private static void insertInitialRows(ISession session) throws Exception { + List schemas = + Collections.singletonList(new MeasurementSchema("s0", TSDataType.INT64)); + Tablet tablet = new Tablet(DEVICE, schemas); + for (int i = 0; i < INITIAL_ROW_COUNT; i++) { + int rowIndex = tablet.getRowSize(); + if (rowIndex == tablet.getMaxRowNumber()) { + session.insertAlignedTablet(tablet); + tablet.reset(); + rowIndex = 0; + } + tablet.addTimestamp(rowIndex, i); + tablet.addValue("s0", rowIndex, (long) i); + } + if (tablet.getRowSize() > 0) { + session.insertAlignedTablet(tablet); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/performance/IoTDBAlignedTVListPrimitiveArrayPerformanceIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/performance/IoTDBAlignedTVListPrimitiveArrayPerformanceIT.java new file mode 100644 index 0000000000000..c0bbf499062e1 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/performance/IoTDBAlignedTVListPrimitiveArrayPerformanceIT.java @@ -0,0 +1,584 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.it.performance; + +import org.apache.iotdb.isession.ISession; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Locale; +import java.util.Properties; + +@RunWith(IoTDBTestRunner.class) +@Category({LocalStandaloneIT.class}) +public class IoTDBAlignedTVListPrimitiveArrayPerformanceIT { + + private static final Logger LOGGER = + LoggerFactory.getLogger(IoTDBAlignedTVListPrimitiveArrayPerformanceIT.class); + + private static final String PROPERTY_PREFIX = "iotdb.aligned.lazy-allocation.perf."; + private static final String ENABLED_PROPERTY = PROPERTY_PREFIX + "enabled"; + private static final String IMPLEMENTATION_PROPERTY = PROPERTY_PREFIX + "implementation"; + private static final String REVISION_PROPERTY = PROPERTY_PREFIX + "revision"; + private static final String HISTORICAL_ROW_COUNT_PROPERTY = + PROPERTY_PREFIX + "historical.row.count"; + private static final String NEW_COLUMN_COUNT_PROPERTY = PROPERTY_PREFIX + "new.column.count"; + private static final String NO_NULL_BATCH_COUNT_PROPERTY = + PROPERTY_PREFIX + "no-null.batch.count"; + private static final String WARMUP_ROUND_COUNT_PROPERTY = PROPERTY_PREFIX + "warmup.round.count"; + private static final String ROUND_COUNT_PROPERTY = PROPERTY_PREFIX + "round.count"; + private static final String RESULT_PATH_PROPERTY = PROPERTY_PREFIX + "result.path"; + private static final String BASELINE_RESULT_PATH_PROPERTY = + PROPERTY_PREFIX + "baseline.result.path"; + private static final String REPORT_PATH_PROPERTY = PROPERTY_PREFIX + "report.path"; + + private static final int DATANODE_MAX_HEAP_SIZE_IN_MB = 1536; + private static final int PRIMITIVE_ARRAY_SIZE = 64; + private static final int HISTORICAL_ROW_COUNT = + Integer.getInteger(HISTORICAL_ROW_COUNT_PROPERTY, 10_000); + private static final int NEW_COLUMN_COUNT = Integer.getInteger(NEW_COLUMN_COUNT_PROPERTY, 512); + private static final int NO_NULL_ROWS_PER_BATCH = PRIMITIVE_ARRAY_SIZE; + private static final int NO_NULL_BATCH_COUNT = + Integer.getInteger(NO_NULL_BATCH_COUNT_PROPERTY, 50); + private static final int WARMUP_ROUND_COUNT = Integer.getInteger(WARMUP_ROUND_COUNT_PROPERTY, 1); + private static final int ROUND_COUNT = Integer.getInteger(ROUND_COUNT_PROPERTY, 5); + private static final String IMPLEMENTATION = + System.getProperty(IMPLEMENTATION_PROPERTY, "candidate"); + private static final String REVISION = System.getProperty(REVISION_PROPERTY, "unknown"); + private static final String DEVICE_PREFIX = "root.aligned_lazy_allocation_performance.d"; + private static final String NO_NULL_DEVICE = "root.aligned_lazy_allocation_performance.no_null"; + + @BeforeClass + public static void setUp() throws Exception { + Assume.assumeTrue(Boolean.getBoolean(ENABLED_PROPERTY)); + EnvFactory.getEnv() + .getConfig() + .getDataNodeJVMConfig() + .setMaxHeapSize(DATANODE_MAX_HEAP_SIZE_IN_MB); + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setAutoCreateSchemaEnabled(false) + .setEnableMemControl(true) + .setPrimitiveArraySize(PRIMITIVE_ARRAY_SIZE) + .setMemtableSizeThreshold(768L * 1024 * 1024) + .setDatanodeMemoryProportion("6:1:1:1:1:1") + .setWriteMemoryProportion("100:1"); + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testAlignedWritePerformance() throws Exception { + Assert.assertTrue(HISTORICAL_ROW_COUNT > 0); + Assert.assertTrue(NEW_COLUMN_COUNT > 0); + Assert.assertTrue(NO_NULL_BATCH_COUNT > 0); + Assert.assertTrue(WARMUP_ROUND_COUNT >= 0); + Assert.assertTrue(ROUND_COUNT > 0); + + List newMeasurements = new ArrayList<>(NEW_COLUMN_COUNT); + List newDataTypes = new ArrayList<>(NEW_COLUMN_COUNT); + List newValues = new ArrayList<>(NEW_COLUMN_COUNT); + for (int i = 1; i <= NEW_COLUMN_COUNT; i++) { + newMeasurements.add("s" + i); + newDataTypes.add(TSDataType.INT64); + newValues.add((long) i); + } + + long[] lazyAllocationElapsedNanos; + long[] noNullElapsedNanos; + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + lazyAllocationElapsedNanos = + runLazyAllocationWorkload(session, newMeasurements, newDataTypes, newValues); + noNullElapsedNanos = runNoNullWorkload(session); + } + + BenchmarkResult result = + new BenchmarkResult( + IMPLEMENTATION, + REVISION, + HISTORICAL_ROW_COUNT, + NEW_COLUMN_COUNT, + WARMUP_ROUND_COUNT, + NO_NULL_BATCH_COUNT, + lazyAllocationElapsedNanos, + noNullElapsedNanos); + Path resultPath = getResultPath(); + writeResult(resultPath, result); + LOGGER.info("AlignedTVList lazy-allocation performance result: {}", result.toSummary()); + LOGGER.info("Performance result written to {}", resultPath.toAbsolutePath()); + + String baselineResultPath = System.getProperty(BASELINE_RESULT_PATH_PROPERTY); + if (baselineResultPath != null && !baselineResultPath.isEmpty()) { + BenchmarkResult baseline = readResult(Paths.get(baselineResultPath)); + assertComparable(baseline, result); + Path reportPath = + Paths.get( + System.getProperty( + REPORT_PATH_PROPERTY, + "target/aligned-tvlist-lazy-allocation-performance-report.md")); + writeFile(reportPath, buildReport(baseline, result)); + LOGGER.info("Comparison report written to {}", reportPath.toAbsolutePath()); + } + } + + private static long[] runLazyAllocationWorkload( + ISession session, + List newMeasurements, + List newDataTypes, + List newValues) + throws Exception { + long[] elapsedNanos = new long[ROUND_COUNT]; + int totalRoundCount = WARMUP_ROUND_COUNT + ROUND_COUNT; + for (int round = 0; round < totalRoundCount; round++) { + String device = DEVICE_PREFIX + round; + createAlignedTimeseries(session, device, NEW_COLUMN_COUNT + 1); + insertHistoricalRows(session, device); + + long startNanos = System.nanoTime(); + session.insertAlignedRecord( + device, HISTORICAL_ROW_COUNT, newMeasurements, newDataTypes, newValues); + long roundElapsedNanos = System.nanoTime() - startNanos; + if (round >= WARMUP_ROUND_COUNT) { + elapsedNanos[round - WARMUP_ROUND_COUNT] = roundElapsedNanos; + } + + assertLazyAllocationDataWritten(session, device); + } + return elapsedNanos; + } + + private static long[] runNoNullWorkload(ISession session) throws Exception { + createAlignedTimeseries(session, NO_NULL_DEVICE, NEW_COLUMN_COUNT); + List schemas = new ArrayList<>(NEW_COLUMN_COUNT); + for (int column = 0; column < NEW_COLUMN_COUNT; column++) { + schemas.add(new MeasurementSchema("s" + column, TSDataType.INT64)); + } + Tablet tablet = new Tablet(NO_NULL_DEVICE, schemas, NO_NULL_ROWS_PER_BATCH); + for (int column = 0; column < NEW_COLUMN_COUNT; column++) { + long[] columnValues = (long[]) tablet.getValues()[column]; + for (int row = 0; row < NO_NULL_ROWS_PER_BATCH; row++) { + columnValues[row] = column + row; + } + } + tablet.setRowSize(NO_NULL_ROWS_PER_BATCH); + + long nextTimestamp = 0; + long[] elapsedNanos = new long[ROUND_COUNT]; + int totalRoundCount = WARMUP_ROUND_COUNT + ROUND_COUNT; + for (int round = 0; round < totalRoundCount; round++) { + long startNanos = System.nanoTime(); + for (int batch = 0; batch < NO_NULL_BATCH_COUNT; batch++) { + for (int row = 0; row < NO_NULL_ROWS_PER_BATCH; row++) { + tablet.getTimestamps()[row] = nextTimestamp++; + } + session.insertAlignedTablet(tablet); + } + long roundElapsedNanos = System.nanoTime() - startNanos; + if (round >= WARMUP_ROUND_COUNT) { + elapsedNanos[round - WARMUP_ROUND_COUNT] = roundElapsedNanos; + } + } + + assertNoNullDataWritten(session, nextTimestamp); + return elapsedNanos; + } + + private static void createAlignedTimeseries(ISession session, String device, int measurementCount) + throws Exception { + StringBuilder statement = new StringBuilder("CREATE ALIGNED TIMESERIES "); + statement.append(device).append("(s0 INT64"); + for (int i = 1; i < measurementCount; i++) { + statement.append(",s").append(i).append(" INT64"); + } + statement.append(')'); + session.executeNonQueryStatement(statement.toString()); + } + + private static void insertHistoricalRows(ISession session, String device) throws Exception { + List schemas = + Collections.singletonList(new MeasurementSchema("s0", TSDataType.INT64)); + Tablet tablet = new Tablet(device, schemas, Math.min(HISTORICAL_ROW_COUNT, 1024)); + for (int row = 0; row < HISTORICAL_ROW_COUNT; row++) { + int rowIndex = tablet.getRowSize(); + if (rowIndex == tablet.getMaxRowNumber()) { + session.insertAlignedTablet(tablet); + tablet.reset(); + rowIndex = 0; + } + tablet.addTimestamp(rowIndex, row); + tablet.addValue("s0", rowIndex, (long) row); + } + if (tablet.getRowSize() > 0) { + session.insertAlignedTablet(tablet); + } + } + + private static void assertLazyAllocationDataWritten(ISession session, String device) + throws Exception { + try (SessionDataSet dataSet = + session.executeQueryStatement( + "SELECT COUNT(s0), COUNT(s" + NEW_COLUMN_COUNT + ") FROM " + device)) { + Assert.assertTrue(dataSet.hasNext()); + List fields = dataSet.next().getFields(); + Assert.assertEquals(HISTORICAL_ROW_COUNT, fields.get(0).getLongV()); + Assert.assertEquals(1, fields.get(1).getLongV()); + Assert.assertFalse(dataSet.hasNext()); + } + } + + private static void assertNoNullDataWritten(ISession session, long expectedRowCount) + throws Exception { + try (SessionDataSet dataSet = + session.executeQueryStatement( + "SELECT COUNT(s0), COUNT(s" + (NEW_COLUMN_COUNT - 1) + ") FROM " + NO_NULL_DEVICE)) { + Assert.assertTrue(dataSet.hasNext()); + List fields = dataSet.next().getFields(); + Assert.assertEquals(expectedRowCount, fields.get(0).getLongV()); + Assert.assertEquals(expectedRowCount, fields.get(1).getLongV()); + Assert.assertFalse(dataSet.hasNext()); + } + } + + private static Path getResultPath() { + String defaultFileName = + "target/aligned-tvlist-lazy-allocation-" + + IMPLEMENTATION.replaceAll("[^A-Za-z0-9_.-]", "_") + + ".properties"; + return Paths.get(System.getProperty(RESULT_PATH_PROPERTY, defaultFileName)); + } + + private static void writeResult(Path path, BenchmarkResult result) throws IOException { + Properties properties = new Properties(); + properties.setProperty("format.version", "2"); + properties.setProperty("implementation", result.implementation); + properties.setProperty("revision", result.revision); + properties.setProperty("historical.row.count", Integer.toString(result.historicalRowCount)); + properties.setProperty("new.column.count", Integer.toString(result.newColumnCount)); + properties.setProperty("warmup.round.count", Integer.toString(result.warmupRoundCount)); + properties.setProperty( + "round.count", Integer.toString(result.lazyAllocationElapsedNanos.length)); + properties.setProperty("no-null.batch.count", Integer.toString(result.noNullBatchCount)); + properties.setProperty("no-null.rows.per.batch", Integer.toString(NO_NULL_ROWS_PER_BATCH)); + properties.setProperty( + "lazy-allocation.latencies.nanos", join(result.lazyAllocationElapsedNanos)); + properties.setProperty("no-null.latencies.nanos", join(result.noNullElapsedNanos)); + createParentDirectories(path); + try (OutputStream outputStream = Files.newOutputStream(path)) { + properties.store(outputStream, "AlignedTVList lazy-allocation integration benchmark"); + } + } + + private static BenchmarkResult readResult(Path path) throws IOException { + Properties properties = new Properties(); + try (InputStream inputStream = Files.newInputStream(path)) { + properties.load(inputStream); + } + return new BenchmarkResult( + properties.getProperty("implementation"), + properties.getProperty("revision"), + Integer.parseInt(properties.getProperty("historical.row.count")), + Integer.parseInt(properties.getProperty("new.column.count")), + Integer.parseInt(properties.getProperty("warmup.round.count")), + Integer.parseInt(properties.getProperty("no-null.batch.count")), + readLongArray(properties, "lazy-allocation.latencies.nanos"), + readLongArray(properties, "no-null.latencies.nanos")); + } + + private static long[] readLongArray(Properties properties, String key) { + String[] values = properties.getProperty(key).split(","); + long[] result = new long[values.length]; + for (int i = 0; i < values.length; i++) { + result[i] = Long.parseLong(values[i]); + } + return result; + } + + private static void assertComparable(BenchmarkResult baseline, BenchmarkResult candidate) { + Assert.assertEquals(baseline.historicalRowCount, candidate.historicalRowCount); + Assert.assertEquals(baseline.newColumnCount, candidate.newColumnCount); + Assert.assertEquals(baseline.warmupRoundCount, candidate.warmupRoundCount); + Assert.assertEquals(baseline.noNullBatchCount, candidate.noNullBatchCount); + Assert.assertEquals( + baseline.lazyAllocationElapsedNanos.length, candidate.lazyAllocationElapsedNanos.length); + Assert.assertEquals(baseline.noNullElapsedNanos.length, candidate.noNullElapsedNanos.length); + } + + private static String buildReport(BenchmarkResult baseline, BenchmarkResult candidate) { + StringBuilder report = new StringBuilder(); + report.append("# AlignedTVList integration performance report\n\n"); + report.append( + "Both revisions were built in separate Git worktrees and ran this same " + + "`LocalStandaloneIT` against a real 1C1D IoTDB environment.\n\n"); + report.append( + String.format( + Locale.ROOT, + "- Lazy-allocation workload: write %,d historical rows only to `s0`, then write " + + "one row to %,d newly used columns.%n" + + "- No-null workload: write %d aligned tablets per round, with %,d columns and " + + "%d rows per tablet; every value is non-null.%n" + + "- Primitive array size: %d%n" + + "- DataNode maximum heap: %,d MiB%n" + + "- Warm-up rounds: %d%n" + + "- Measured rounds: %d%n%n", + candidate.historicalRowCount, + candidate.newColumnCount, + candidate.noNullBatchCount, + candidate.newColumnCount, + NO_NULL_ROWS_PER_BATCH, + PRIMITIVE_ARRAY_SIZE, + DATANODE_MAX_HEAP_SIZE_IN_MB, + candidate.warmupRoundCount, + candidate.lazyAllocationElapsedNanos.length)); + report.append("| Workload | Revision | Commit | Median ms | P95 ms | Values/s |\n"); + report.append("| --- | --- | --- | ---: | ---: | ---: |\n"); + appendReportRow(report, "Lazy allocation", baseline, false); + appendReportRow(report, "Lazy allocation", candidate, false); + appendReportRow(report, "No nulls", baseline, true); + appendReportRow(report, "No nulls", candidate, true); + report.append('\n'); + report.append( + String.format( + Locale.ROOT, + "- Lazy-allocation median latency improvement: %.2f%%%n" + + "- Lazy-allocation throughput improvement: %.2f%%%n" + + "- No-null median latency improvement: %.2f%%%n" + + "- No-null throughput improvement: %.2f%%%n%n", + latencyImprovement(baseline.lazyMedianNanos(), candidate.lazyMedianNanos()), + throughputImprovement(baseline.lazyValuesPerSecond(), candidate.lazyValuesPerSecond()), + latencyImprovement(baseline.noNullMedianNanos(), candidate.noNullMedianNanos()), + throughputImprovement( + baseline.noNullValuesPerSecond(), candidate.noNullValuesPerSecond()))); + report.append( + String.format( + Locale.ROOT, + "- %s lazy-allocation latencies (ms): `%s`%n" + + "- %s lazy-allocation latencies (ms): `%s`%n" + + "- %s no-null latencies (ms): `%s`%n" + + "- %s no-null latencies (ms): `%s`%n%n", + baseline.implementation, + latenciesMillis(baseline.lazyAllocationElapsedNanos), + candidate.implementation, + latenciesMillis(candidate.lazyAllocationElapsedNanos), + baseline.implementation, + latenciesMillis(baseline.noNullElapsedNanos), + candidate.implementation, + latenciesMillis(candidate.noNullElapsedNanos))); + report.append( + "The timed regions contain the client calls and complete server write path; schema " + + "creation, workload preparation, and result verification are outside them. Raw " + + "per-round latencies are retained in the generated result property files.\n\n"); + report.append("## Reproduction\n\n"); + report.append( + "1. Create the baseline worktree: `git worktree add --detach master`.\n" + + "2. Copy this test source to the identical path in the baseline worktree.\n" + + "3. In each worktree, build with " + + "`mvn -Pwith-integration-tests -pl integration-test -am package -DskipTests`.\n" + + String.format( + Locale.ROOT, + "4. Run the baseline with `-D%s=true`, `-D%s=master`, " + + "`-D%s=`, and `-D%s=`.%n" + + "5. Run the candidate with the corresponding candidate values plus " + + "`-D%s=` and `-D%s=`. For both runs, select " + + "this class with `-Dit.test=IoTDBAlignedTVListPrimitiveArrayPerformanceIT` " + + "and execute `failsafe:integration-test@integration-test " + + "failsafe:verify@verify`.%n", + ENABLED_PROPERTY, + IMPLEMENTATION_PROPERTY, + REVISION_PROPERTY, + RESULT_PATH_PROPERTY, + BASELINE_RESULT_PATH_PROPERTY, + REPORT_PATH_PROPERTY)); + return report.toString(); + } + + private static void appendReportRow( + StringBuilder report, String workload, BenchmarkResult result, boolean noNull) { + long medianNanos = noNull ? result.noNullMedianNanos() : result.lazyMedianNanos(); + long p95Nanos = noNull ? result.noNullP95Nanos() : result.lazyP95Nanos(); + double valuesPerSecond = noNull ? result.noNullValuesPerSecond() : result.lazyValuesPerSecond(); + report.append( + String.format( + Locale.ROOT, + "| %s | %s | `%s` | %.3f | %.3f | %.0f |%n", + workload, + result.implementation, + result.revision, + medianNanos / 1_000_000.0, + p95Nanos / 1_000_000.0, + valuesPerSecond)); + } + + private static double latencyImprovement(double baseline, double candidate) { + return (baseline - candidate) * 100.0 / baseline; + } + + private static double throughputImprovement(double baseline, double candidate) { + return (candidate - baseline) * 100.0 / baseline; + } + + private static String latenciesMillis(long[] elapsedNanos) { + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < elapsedNanos.length; i++) { + if (i > 0) { + builder.append(", "); + } + builder.append(String.format(Locale.ROOT, "%.3f", elapsedNanos[i] / 1_000_000.0)); + } + return builder.toString(); + } + + private static void writeFile(Path path, String content) throws IOException { + createParentDirectories(path); + Files.writeString(path, content, StandardCharsets.UTF_8); + } + + private static void createParentDirectories(Path path) throws IOException { + Path parent = path.toAbsolutePath().getParent(); + if (parent != null) { + Files.createDirectories(parent); + } + } + + private static String join(long[] values) { + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < values.length; i++) { + if (i > 0) { + builder.append(','); + } + builder.append(values[i]); + } + return builder.toString(); + } + + private static class BenchmarkResult { + private final String implementation; + private final String revision; + private final int historicalRowCount; + private final int newColumnCount; + private final int warmupRoundCount; + private final int noNullBatchCount; + private final long[] lazyAllocationElapsedNanos; + private final long[] noNullElapsedNanos; + + private BenchmarkResult( + String implementation, + String revision, + int historicalRowCount, + int newColumnCount, + int warmupRoundCount, + int noNullBatchCount, + long[] lazyAllocationElapsedNanos, + long[] noNullElapsedNanos) { + this.implementation = implementation; + this.revision = revision; + this.historicalRowCount = historicalRowCount; + this.newColumnCount = newColumnCount; + this.warmupRoundCount = warmupRoundCount; + this.noNullBatchCount = noNullBatchCount; + this.lazyAllocationElapsedNanos = lazyAllocationElapsedNanos; + this.noNullElapsedNanos = noNullElapsedNanos; + } + + private static long medianNanos(long[] elapsedNanos) { + long[] sorted = Arrays.copyOf(elapsedNanos, elapsedNanos.length); + Arrays.sort(sorted); + return sorted[sorted.length / 2]; + } + + private static long p95Nanos(long[] elapsedNanos) { + long[] sorted = Arrays.copyOf(elapsedNanos, elapsedNanos.length); + Arrays.sort(sorted); + return sorted[(int) Math.ceil(sorted.length * 0.95) - 1]; + } + + private long lazyMedianNanos() { + return medianNanos(lazyAllocationElapsedNanos); + } + + private long lazyP95Nanos() { + return p95Nanos(lazyAllocationElapsedNanos); + } + + private long noNullMedianNanos() { + return medianNanos(noNullElapsedNanos); + } + + private long noNullP95Nanos() { + return p95Nanos(noNullElapsedNanos); + } + + private double lazyValuesPerSecond() { + return newColumnCount * 1_000_000_000.0 / lazyMedianNanos(); + } + + private double noNullValuesPerSecond() { + return (double) newColumnCount + * NO_NULL_ROWS_PER_BATCH + * noNullBatchCount + * 1_000_000_000.0 + / noNullMedianNanos(); + } + + private String toSummary() { + return String.format( + Locale.ROOT, + "implementation=%s, revision=%s, lazy median=%.3f ms, lazy values/s=%.0f, " + + "no-null median=%.3f ms, no-null values/s=%.0f", + implementation, + revision, + lazyMedianNanos() / 1_000_000.0, + lazyValuesPerSecond(), + noNullMedianNanos() / 1_000_000.0, + noNullValuesPerSecond()); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java index b0d00562a4c75..05fd1d4073c2a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java @@ -132,6 +132,10 @@ public boolean containsMeasurement(String measurementId) { return measurementIndexMap.containsKey(measurementId); } + int getMeasurementIndex(String measurementId) { + return measurementIndexMap.get(measurementId); + } + @Override public void putLong(long t, long v) { throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE + TSDataType.VECTOR); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 4528b44f59894..8b14459a7600e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -92,6 +92,7 @@ import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.read.filter.basic.Filter; import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.write.writer.RestorableTsFileIOWriter; import org.slf4j.Logger; @@ -104,9 +105,11 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.CopyOnWriteArrayList; @@ -563,6 +566,7 @@ private long[] checkAlignedMemCost( insertTabletNode.getMeasurements(), insertTabletNode.getDataTypes(), insertTabletNode.getColumns(), + insertTabletNode.getBitMaps(), insertTabletNode.getColumnCategories(), splitStart, splitEnd, @@ -714,9 +718,11 @@ private long[] checkMemCostAndAddToTspInfoForRow( Object[] values, TsTableColumnCategory[] columnCategories) throws WriteProcessException { - // Memory of increased PrimitiveArray and TEXT values, e.g., add a long[128], add 128*8 + // Fixed-size TVList structures and materialized value primitive arrays. long memTableIncrement = 0L; + // Variable-length payload retained by Binary values. long textDataIncrement = 0L; + // ChunkMetadata entries created for newly introduced writable fields. long chunkMetadataIncrement = 0L; for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { @@ -808,9 +814,11 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRow( Object[] values, TsTableColumnCategory[] columnCategories) throws WriteProcessException { - // Memory of increased PrimitiveArray and TEXT values, e.g., add a long[128], add 128*8 + // Fixed-size TVList structures and materialized value primitive arrays. long memTableIncrement = 0L; + // Variable-length payload retained by Binary values. long textDataIncrement = 0L; + // ChunkMetadata entries created for newly introduced writable fields. long chunkMetadataIncrement = 0L; IWritableMemChunk memChunk = @@ -818,46 +826,61 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRow( if (memChunk == null) { TSDataType[] writableFieldDataTypes = getWritableFieldDataTypes(measurements, dataTypes, values, columnCategories); - // For new device of this mem table - // ChunkMetadataIncrement + // A new aligned device creates one value ChunkMetadata entry for each writable field. chunkMetadataIncrement += ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER, TSDataType.VECTOR) * writableFieldDataTypes.length; + // The first row creates the first aligned TVList block. All writable fields have a non-null + // value, so this includes the timestamp array, value primitive arrays, bitmap reservations, + // array headers, and ArrayList references for that block. memTableIncrement += AlignedTVList.alignedTvListArrayMemCost(writableFieldDataTypes, null); } else { // For existed device of this mem table AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk; - List dataTypesInTVList = new ArrayList<>(); + int currentPointNum = alignedMemChunk.alignedListSize(); + int currentArrayNum = + currentPointNum / PrimitiveArrayManager.ARRAY_SIZE + + (currentPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); + int targetArrayIndex = currentPointNum / PrimitiveArrayManager.ARRAY_SIZE; + int newColumnCount = 0; for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { // Skip failed Measurements if (!isWritableFieldMeasurement(measurements, dataTypes, values, columnCategories, i)) { continue; } - // add arrays for new columns if (!alignedMemChunk.containsMeasurement(measurements[i])) { - int currentArrayNum = - alignedMemChunk.alignedListSize() / PrimitiveArrayManager.ARRAY_SIZE - + (alignedMemChunk.alignedListSize() % PrimitiveArrayManager.ARRAY_SIZE > 0 - ? 1 - : 0); - memTableIncrement += currentArrayNum * AlignedTVList.valueListArrayMemCost(dataTypes[i]); - dataTypesInTVList.add(dataTypes[i]); + // Extending a column adds one null value-list placeholder and one conservatively reserved + // bitmap to every historical block. No value primitive array is allocated there. + memTableIncrement += + currentArrayNum * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); + newColumnCount++; } - } - // this insertion will result in a new array - if ((alignedMemChunk.alignedListSize() % PrimitiveArrayManager.ARRAY_SIZE) == 0) { - memTableIncrement += alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost(); - for (TSDataType dataType : dataTypesInTVList) { - memTableIncrement += AlignedTVList.valueListArrayMemCost(dataType); + if (!isValueArrayMaterialized(alignedMemChunk, measurements[i], targetArrayIndex)) { + // The non-null value materializes this column's primitive array in the target block. + // This cost contains the primitive-array payload and its array header. + memTableIncrement += AlignedTVList.primitiveArrayMemCost(dataTypes[i]); } } + if ((currentPointNum % PrimitiveArrayManager.ARRAY_SIZE) == 0) { + // Starting a block allocates its timestamp array, optional sort-index array, value-list + // null + // placeholders, bitmap reservations, array headers, and ArrayList references for all + // existing columns. Value primitive arrays are excluded because they were charged above + // only for written columns. + memTableIncrement += + alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCostWithoutPrimitiveArrays(); + // The working TVList does not contain newly extended columns yet, so add their value-list + // placeholder and bitmap reservation for this new block separately. + memTableIncrement += + (long) newColumnCount * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); + } } for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { - // TEXT data mem size if (isWritableFieldMeasurement(measurements, dataTypes, values, columnCategories, i) && dataTypes[i].isBinary()) { + // Binary values keep their variable-length content outside the fixed Binary[] array. textDataIncrement += MemUtils.getBinarySize((Binary) values[i]); } } @@ -868,12 +891,15 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRow( @SuppressWarnings("squid:S3776") // high Cognitive Complexity private long[] checkAlignedMemCostAndAddToTspInfoForRows(List insertRowNodeList) throws WriteProcessException { - // Memory of increased PrimitiveArray and TEXT values, e.g., add a long[128], add 128*8 + // Fixed-size TVList structures and materialized value primitive arrays. long memTableIncrement = 0L; + // Variable-length payload retained by Binary values. long textDataIncrement = 0L; + // ChunkMetadata entries created for newly introduced writable fields. long chunkMetadataIncrement = 0L; // device -> (measurements -> datatype, adding aligned TVList size) Map, Integer>> increasingMemTableInfo = new HashMap<>(); + Map>> materializedArraysInCurrentBatch = new HashMap<>(); for (InsertRowNode insertRowNode : insertRowNodeList) { IDeviceID deviceId = insertRowNode.getDeviceID(); TSDataType[] dataTypes = insertRowNode.getDataTypes(); @@ -888,11 +914,13 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins measurements, dataTypes, values, insertRowNode.getColumnCategories()); Pair, Integer> addingPointNumInfo = increasingMemTableInfo.computeIfAbsent(deviceId, k -> new Pair<>(new HashMap<>(), 0)); - // For new device of this mem table - // ChunkMetadataIncrement + // The first row of a new aligned device creates one value ChunkMetadata entry for each + // writable field. chunkMetadataIncrement += ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER, TSDataType.VECTOR) * writableFieldDataTypes.length; + // The first row creates the first complete aligned TVList block: timestamp array, value + // primitive arrays for the writable fields, bitmap reservations, headers, and references. memTableIncrement += AlignedTVList.alignedTvListArrayMemCost(writableFieldDataTypes, null); for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { // Skip failed Measurements @@ -901,6 +929,10 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins continue; } addingPointNumInfo.left.put(measurements[i], dataTypes[i]); + materializedArraysInCurrentBatch + .computeIfAbsent(deviceId, key -> new HashMap<>()) + .computeIfAbsent(measurements[i], key -> new HashSet<>()) + .add(0); } addingPointNumInfo.setRight(1); @@ -908,9 +940,11 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins // For existed device of this mem table AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk; int currentChunkPointNum = alignedMemChunk == null ? 0 : alignedMemChunk.alignedListSize(); - List dataTypesInTVList = new ArrayList<>(); Pair, Integer> addingPointNumInfo = increasingMemTableInfo.computeIfAbsent(deviceId, k -> new Pair<>(new HashMap<>(), 0)); + int addingPointNum = addingPointNumInfo.getRight(); + int pointNumBeforeCurrentRow = currentChunkPointNum + addingPointNum; + int targetArrayIndex = pointNumBeforeCurrentRow / PrimitiveArrayManager.ARRAY_SIZE; for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { // Skip failed Measurements if (!isWritableFieldMeasurement( @@ -918,7 +952,6 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins continue; } - int addingPointNum = addingPointNumInfo.getRight(); // Extending the column of aligned mem chunk boolean currentMemChunkContainsMeasurement = alignedMemChunk != null && alignedMemChunk.containsMeasurement(measurements[i]); @@ -926,27 +959,46 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins && !addingPointNumInfo.left.containsKey(measurements[i])) { addingPointNumInfo.left.put(measurements[i], dataTypes[i]); int currentArrayNum = - (currentChunkPointNum + addingPointNum) / PrimitiveArrayManager.ARRAY_SIZE - + ((currentChunkPointNum + addingPointNum) % PrimitiveArrayManager.ARRAY_SIZE - > 0 - ? 1 - : 0); + pointNumBeforeCurrentRow / PrimitiveArrayManager.ARRAY_SIZE + + (pointNumBeforeCurrentRow % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); + // A column first seen in this batch adds a null value-list placeholder and a bitmap + // reservation to every block that already exists before the current row. memTableIncrement += - currentArrayNum * AlignedTVList.valueListArrayMemCost(dataTypes[i]); + currentArrayNum * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); + } + Set materializedArrayIndexes = + materializedArraysInCurrentBatch + .computeIfAbsent(deviceId, key -> new HashMap<>()) + .computeIfAbsent(measurements[i], key -> new HashSet<>()); + if (!isValueArrayMaterialized(alignedMemChunk, measurements[i], targetArrayIndex) + && materializedArrayIndexes.add(targetArrayIndex)) { + // Charge the payload and header of a value primitive array exactly once when this + // batch first writes a non-null value to the column in the target block. + memTableIncrement += AlignedTVList.primitiveArrayMemCost(dataTypes[i]); } } - int addingPointNum = addingPointNumInfo.right; - // Here currentChunkPointNum + addingPointNum >= 1 - if (((currentChunkPointNum + addingPointNum) % PrimitiveArrayManager.ARRAY_SIZE) == 0) { - dataTypesInTVList.addAll(addingPointNumInfo.left.values()); - memTableIncrement += - alignedMemChunk != null - ? alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost() - + dataTypesInTVList.stream() - .mapToLong(AlignedTVList::valueListArrayMemCost) - .sum() - : AlignedTVList.alignedTvListArrayMemCost( - dataTypesInTVList.toArray(new TSDataType[0]), null); + if ((pointNumBeforeCurrentRow % PrimitiveArrayManager.ARRAY_SIZE) == 0) { + if (alignedMemChunk == null) { + // A device created earlier in this batch starts another block. Charge the timestamp + // array, value-list placeholders, bitmap reservations, headers, and references for all + // columns discovered in the batch, excluding their value primitive arrays. + memTableIncrement += + AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays( + addingPointNumInfo.left.values().toArray(new TSDataType[0]), null); + } else { + // An existing aligned TVList starts another block. Charge its timestamp array, optional + // sort-index array, value-list placeholders, bitmap reservations, headers, and + // references for columns that were already present before this batch. + memTableIncrement += + alignedMemChunk + .getWorkingTVList() + .alignedTvListArrayMemCostWithoutPrimitiveArrays(); + // Columns first introduced by this batch are absent from the working TVList's type + // list, so add one value-list placeholder and bitmap reservation for each of them. + memTableIncrement += + (long) addingPointNumInfo.left.size() + * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); + } } addingPointNumInfo.setRight(addingPointNum + 1); } @@ -957,8 +1009,8 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins measurements, dataTypes, values, insertRowNode.getColumnCategories(), i)) { continue; } - // TEXT data mem size if (dataTypes[i].isBinary() && values[i] != null) { + // Binary values keep their variable-length content outside the fixed Binary[] array. textDataIncrement += MemUtils.getBinarySize((Binary) values[i]); } } @@ -1000,6 +1052,7 @@ private long[] checkAlignedMemCostAndAddToTspForTablet( String[] measurements, TSDataType[] dataTypes, Object[] columns, + BitMap[] bitMaps, TsTableColumnCategory[] columnCategories, int start, int end, @@ -1018,6 +1071,7 @@ private long[] checkAlignedMemCostAndAddToTspForTablet( end, memIncrements, columns, + bitMaps, columnCategories, results); long memTableIncrement = memIncrements[0]; @@ -1035,7 +1089,9 @@ private void updateMemCost( int end, long[] memIncrements, Object column) { - // memIncrements = [memTable, text, chunk metadata] respectively + // memIncrements[0]: fixed-size TVList structures and materialized value primitive arrays. + // memIncrements[1]: variable-length payload retained by Binary values. + // memIncrements[2]: ChunkMetadata entries for newly introduced writable fields. IWritableMemChunk memChunk = workMemTable.getWritableMemChunk(deviceId, measurement); if (memChunk == null) { @@ -1074,6 +1130,7 @@ private void updateAlignedMemCost( int end, long[] memIncrements, Object[] columns, + BitMap[] bitMaps, TsTableColumnCategory[] columnCategories, TSStatus[] results) { int incomingPointNum = end - start; @@ -1085,21 +1142,38 @@ private void updateAlignedMemCost( IWritableMemChunk memChunk = workMemTable.getWritableMemChunk(deviceId, AlignedPath.VECTOR_PLACEHOLDER); if (memChunk == null) { - // new devices introduce new ChunkMetadata - // ChunkMetadata memory Increment + // A new aligned device creates one value ChunkMetadata entry for each writable field. memIncrements[2] += writableFieldDataTypes.length * ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER, TSDataType.VECTOR); - // TVList memory int numArraysToAdd = incomingPointNum / PrimitiveArrayManager.ARRAY_SIZE + (incomingPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); + // Each new block allocates its timestamp array, value-list null placeholders, bitmap + // reservations, array headers, and ArrayList references. Value primitive arrays are excluded + // here because all-null and failed-only column segments leave them unmaterialized. + memIncrements[0] += + numArraysToAdd + * AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays( + writableFieldDataTypes, null); + // Add the payload and header of a value primitive array only for a column/block segment that + // contains at least one successful non-null value. memIncrements[0] += - numArraysToAdd * AlignedTVList.alignedTvListArrayMemCost(writableFieldDataTypes, null); + calculateTabletPrimitiveArrayMemCost( + null, + measurementIds, + dataTypes, + columns, + bitMaps, + columnCategories, + results, + start, + end, + 0); } else { AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk; - List dataTypesInTVList = new ArrayList<>(); + List newDataTypes = new ArrayList<>(); int currentPointNum = alignedMemChunk.alignedListSize(); int newPointNum = currentPointNum + incomingPointNum; int currentArrayCnt = @@ -1112,9 +1186,12 @@ private void updateAlignedMemCost( } if (!alignedMemChunk.containsMeasurement(measurementIds[i])) { - // add a new column in the TVList, the new column should be as long as existing ones - memIncrements[0] += currentArrayCnt * AlignedTVList.valueListArrayMemCost(dataType); - dataTypesInTVList.add(dataType); + // A new column adds one null value-list placeholder and one conservatively reserved + // bitmap to every historical block. It does not materialize value primitive arrays in + // those blocks. + memIncrements[0] += + currentArrayCnt * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); + newDataTypes.add(dataType); } } @@ -1125,13 +1202,36 @@ private void updateAlignedMemCost( long acquireArray = newArrayCnt - currentArrayCnt; if (acquireArray != 0) { - // memory of extending the TVList + // Each acquired block adds a timestamp array, optional sort-index array, value-list null + // placeholders, bitmap reservations, array headers, and ArrayList references for columns + // already present in the working TVList. Value primitive arrays are charged separately + // below. memIncrements[0] += - acquireArray * alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost(); - for (TSDataType dataType : dataTypesInTVList) { - memIncrements[0] += acquireArray * AlignedTVList.valueListArrayMemCost(dataType); + acquireArray + * alignedMemChunk + .getWorkingTVList() + .alignedTvListArrayMemCostWithoutPrimitiveArrays(); + for (TSDataType ignored : newDataTypes) { + // Newly extended columns are not included in the working TVList's per-block cost yet, so + // add their value-list placeholder and bitmap reservation to every acquired block. + memIncrements[0] += + acquireArray * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); } } + // Charge the payload and header of each value primitive array that this tablet actually + // materializes. Arrays already present in the working TVList are not charged again. + memIncrements[0] += + calculateTabletPrimitiveArrayMemCost( + alignedMemChunk, + measurementIds, + dataTypes, + columns, + bitMaps, + columnCategories, + results, + start, + end, + currentPointNum); } // flexible-length data size @@ -1143,11 +1243,92 @@ private void updateAlignedMemCost( if (dataType.isBinary()) { Binary[] binColumn = (Binary[]) columns[i]; + // Binary values keep their variable-length content outside the fixed Binary[] arrays. + // Failed rows are excluded because their values are not inserted into the memtable. memIncrements[1] += MemUtils.getBinaryColumnSize(binColumn, start, end, results); } } } + /** + * Calculates only the value primitive arrays that an aligned tablet will materialize. The + * per-block timestamp arrays, value-list placeholders, bitmap reservations, headers, and + * references are charged by the caller. A column/block pair is charged only when the tablet has + * at least one successful non-null value in that block and the working TVList has not already + * allocated its value array. + */ + private static long calculateTabletPrimitiveArrayMemCost( + AlignedWritableMemChunk alignedMemChunk, + String[] measurementIds, + TSDataType[] dataTypes, + Object[] columns, + BitMap[] bitMaps, + TsTableColumnCategory[] columnCategories, + TSStatus[] results, + int start, + int end, + int currentPointNum) { + long size = 0; + for (int column = 0; dataTypes != null && column < dataTypes.length; column++) { + if (!isWritableFieldMeasurement( + measurementIds, dataTypes, columns, columnCategories, column)) { + continue; + } + BitMap bitMap = bitMaps == null || column >= bitMaps.length ? null : bitMaps[column]; + int inputIndex = start; + int targetArrayIndex = currentPointNum / PrimitiveArrayManager.ARRAY_SIZE; + int targetElementIndex = currentPointNum % PrimitiveArrayManager.ARRAY_SIZE; + while (inputIndex < end) { + // Map the next tablet segment onto one target TVList block. The first segment may fill the + // unused tail of the current block; subsequent segments start at block offset zero. + int length = + Math.min(end - inputIndex, PrimitiveArrayManager.ARRAY_SIZE - targetElementIndex); + if (containsNonNullValue(bitMap, results, inputIndex, length) + && !isValueArrayMaterialized( + alignedMemChunk, measurementIds[column], targetArrayIndex)) { + // One successful non-null write materializes the whole fixed-size value array, whose + // memory consists of the primitive payload and the array header. + size += AlignedTVList.primitiveArrayMemCost(dataTypes[column]); + } + inputIndex += length; + targetArrayIndex++; + targetElementIndex = 0; + } + } + return size; + } + + private static boolean isValueArrayMaterialized( + AlignedWritableMemChunk alignedMemChunk, String measurement, int arrayIndex) { + if (alignedMemChunk == null || !alignedMemChunk.containsMeasurement(measurement)) { + return false; + } + List valueArrays = + alignedMemChunk + .getWorkingTVList() + .getValues() + .get(alignedMemChunk.getMeasurementIndex(measurement)); + return arrayIndex < valueArrays.size() && valueArrays.get(arrayIndex) != null; + } + + private static boolean containsNonNullValue( + BitMap bitMap, TSStatus[] results, int start, int length) { + if (bitMap == null && results == null) { + return true; + } + for (int i = start; i < start + length; i++) { + boolean isNull = bitMap != null && bitMap.isMarked(i); + boolean isFailed = + results != null + && results[i] != null + && results[i].code != TSStatusCode.SUCCESS_STATUS.getStatusCode(); + if (!isNull && !isFailed) { + return true; + } + } + return false; + } + private static TSDataType[] getWritableFieldDataTypes( String[] measurementIds, TSDataType[] dataTypes, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java index 464cbe09995ed..644890d0bed32 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java @@ -81,6 +81,10 @@ public abstract class AlignedTVList extends TVList { // Record total memory size of binary column protected long[] memoryBinaryChunkSize; + // Number and memory cost of primitive arrays that have been allocated for each value column + private int[] materializedValueArrayCounts; + private long materializedValueArrayMemCost; + // Data type list -> list of TVList, add 1 when expanded -> primitive array of basic type // Index relation: columnIndex(dataTypeIndex) -> arrayIndex -> elementIndex protected List> values; @@ -103,6 +107,7 @@ public abstract class AlignedTVList extends TVList { super(); dataTypes = types; memoryBinaryChunkSize = new long[dataTypes.size()]; + materializedValueArrayCounts = new int[dataTypes.size()]; values = new ArrayList<>(types.size()); for (int i = 0; i < types.size(); i++) { @@ -125,7 +130,9 @@ private List convertToType(TSDataType to, TSDataType from, List if (!to.isCompatible(from)) { return null; } - return originalValues.stream().map(o -> to.castFromArray(from, o)).collect(Collectors.toList()); + return originalValues.stream() + .map(o -> o == null ? null : to.castFromArray(from, o)) + .collect(Collectors.toList()); } @Override @@ -167,6 +174,15 @@ public TVList getTvListByColumnIndex( alignedTvList.allValueColDeletedMap = ignoreAllNullRows ? getAllValueColDeletedMap() : null; alignedTvList.timeColDeletedMap = this.timeColDeletedMap; alignedTvList.timeDeletedCnt = this.timeDeletedCnt; + for (int i = 0; i < columnIndexList.size(); i++) { + int columnIndex = columnIndexList.get(i); + if (columnIndex != -1 && values.get(i) != null) { + int materializedArrayCount = materializedValueArrayCounts[columnIndex]; + alignedTvList.materializedValueArrayCounts[i] = materializedArrayCount; + alignedTvList.materializedValueArrayMemCost += + (long) materializedArrayCount * primitiveArrayMemCost(dataTypeList.get(i)); + } + } return alignedTvList; } @@ -180,6 +196,9 @@ public synchronized AlignedTVList cloneForFlushSort() { cloneList.values = this.values; cloneList.bitMaps = this.bitMaps; cloneList.timeColDeletedMap = this.timeColDeletedMap; + cloneList.materializedValueArrayCounts = + Arrays.copyOf(materializedValueArrayCounts, materializedValueArrayCounts.length); + cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost; return cloneList; } @@ -215,6 +234,9 @@ public synchronized AlignedTVList clone() { } } cloneList.timeColDeletedMap = timeColDeletedMap == null ? null : timeColDeletedMap.clone(); + cloneList.materializedValueArrayCounts = + Arrays.copyOf(materializedValueArrayCounts, materializedValueArrayCounts.length); + cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost; return cloneList; } @@ -229,43 +251,35 @@ public synchronized void putAlignedValue(long timestamp, Object[] value) { timestamps.get(arrayIndex)[elementIndex] = timestamp; for (int i = 0; i < values.size(); i++) { Object columnValue = value[i]; - List columnValues = values.get(i); if (columnValue == null) { markNullValue(i, arrayIndex, elementIndex); + continue; } + Object valueArray = getOrCreateValueArray(i, arrayIndex); switch (dataTypes.get(i)) { case TEXT: case BLOB: case STRING: case OBJECT: - ((Binary[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (Binary) columnValue : Binary.EMPTY_VALUE; - memoryBinaryChunkSize[i] += - columnValue != null - ? getBinarySize((Binary) columnValue) - : getBinarySize(Binary.EMPTY_VALUE); + ((Binary[]) valueArray)[elementIndex] = (Binary) columnValue; + memoryBinaryChunkSize[i] += getBinarySize((Binary) columnValue); break; case FLOAT: - ((float[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (float) columnValue : Float.MIN_VALUE; + ((float[]) valueArray)[elementIndex] = (float) columnValue; break; case INT32: case DATE: - ((int[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (int) columnValue : Integer.MIN_VALUE; + ((int[]) valueArray)[elementIndex] = (int) columnValue; break; case INT64: case TIMESTAMP: - ((long[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (long) columnValue : Long.MIN_VALUE; + ((long[]) valueArray)[elementIndex] = (long) columnValue; break; case DOUBLE: - ((double[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (double) columnValue : Double.MIN_VALUE; + ((double[]) valueArray)[elementIndex] = (double) columnValue; break; case BOOLEAN: - ((boolean[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null && (boolean) columnValue; + ((boolean[]) valueArray)[elementIndex] = (boolean) columnValue; break; default: break; @@ -396,33 +410,7 @@ public void extendColumn(TSDataType dataType) { List columnValue = new ArrayList<>(timestamps.size()); List columnBitMaps = new ArrayList<>(timestamps.size()); for (int i = 0; i < timestamps.size(); i++) { - switch (dataType) { - case TEXT: - case STRING: - case BLOB: - case OBJECT: - columnValue.add(getPrimitiveArraysByType(TSDataType.TEXT)); - break; - case FLOAT: - columnValue.add(getPrimitiveArraysByType(TSDataType.FLOAT)); - break; - case INT32: - case DATE: - columnValue.add(getPrimitiveArraysByType(TSDataType.INT32)); - break; - case INT64: - case TIMESTAMP: - columnValue.add(getPrimitiveArraysByType(TSDataType.INT64)); - break; - case DOUBLE: - columnValue.add(getPrimitiveArraysByType(TSDataType.DOUBLE)); - break; - case BOOLEAN: - columnValue.add(getPrimitiveArraysByType(TSDataType.BOOLEAN)); - break; - default: - break; - } + columnValue.add(null); BitMap bitMap = BitMap.createBitMapDynamically(ARRAY_SIZE); // The following code is for these 2 kinds of scenarios. @@ -448,6 +436,7 @@ public void extendColumn(TSDataType dataType) { memoryBinaryChunkSize = new long[dataTypes.size()]; System.arraycopy( tmpValueChunkRawSize, 0, memoryBinaryChunkSize, 0, tmpValueChunkRawSize.length); + materializedValueArrayCounts = Arrays.copyOf(materializedValueArrayCounts, dataTypes.size()); } private Object getObjectByValueIndex(int rowIndex, int columnIndex) { @@ -621,12 +610,15 @@ public boolean isNullValue(int unsortedRowIndex, int columnIndex) { if (columnIndex < 0 || columnIndex >= values.size() || values.get(columnIndex) == null) { return true; } + int arrayIndex = unsortedRowIndex / ARRAY_SIZE; + if (values.get(columnIndex).get(arrayIndex) == null) { + return true; + } if (bitMaps == null || bitMaps.get(columnIndex) == null - || bitMaps.get(columnIndex).get(unsortedRowIndex / ARRAY_SIZE) == null) { + || bitMaps.get(columnIndex).get(arrayIndex) == null) { return false; } - int arrayIndex = unsortedRowIndex / ARRAY_SIZE; int elementIndex = unsortedRowIndex % ARRAY_SIZE; List columnBitMaps = bitMaps.get(columnIndex); return columnBitMaps.get(arrayIndex).isMarked(elementIndex); @@ -744,6 +736,9 @@ public void deleteColumn(int columnIndex) { } protected Object cloneValue(TSDataType type, Object value) { + if (value == null) { + return null; + } switch (type) { case TEXT: case BLOB: @@ -791,12 +786,16 @@ protected void clearValue() { List columnValues = values.get(i); if (columnValues != null) { for (Object dataArray : columnValues) { - PrimitiveArrayManager.release(dataArray); + if (dataArray != null) { + PrimitiveArrayManager.release(dataArray); + } } columnValues.clear(); } memoryBinaryChunkSize[i] = 0; } + Arrays.fill(materializedValueArrayCounts, 0); + materializedValueArrayMemCost = 0; } @Override @@ -817,7 +816,7 @@ protected void expandValues() { indices.add((int[]) getPrimitiveArraysByType(TSDataType.INT32)); } for (int i = 0; i < dataTypes.size(); i++) { - values.get(i).add(getPrimitiveArraysByType(dataTypes.get(i))); + values.get(i).add(null); if (bitMaps != null && bitMaps.get(i) != null) { bitMaps.get(i).add(null); } @@ -887,7 +886,7 @@ public synchronized void putAlignedValues( if (internalRemaining >= inputRemaining) { // the remaining inputs can fit the last array, copy all remaining inputs into last array System.arraycopy(time, idx, timestamps.get(arrayIdx), elementIdx, inputRemaining); - arrayCopy(value, idx, arrayIdx, elementIdx, inputRemaining); + arrayCopy(value, bitMaps, results, idx, arrayIdx, elementIdx, inputRemaining); for (int i = 0; i < inputRemaining; i++) { if (indices != null) { indices.get(arrayIdx)[elementIdx + i] = rowCount; @@ -900,7 +899,7 @@ public synchronized void putAlignedValues( // the remaining inputs cannot fit the last array, fill the last array and create a new // one and enter the next loop System.arraycopy(time, idx, timestamps.get(arrayIdx), elementIdx, internalRemaining); - arrayCopy(value, idx, arrayIdx, elementIdx, internalRemaining); + arrayCopy(value, bitMaps, results, idx, arrayIdx, elementIdx, internalRemaining); for (int i = 0; i < internalRemaining; i++) { if (indices != null) { indices.get(arrayIdx)[elementIdx + i] = rowCount; @@ -1007,18 +1006,25 @@ private static BitMap buildResultBitMap(TSStatus[] results, int idx, int element return bitmap; } - private void arrayCopy(Object[] value, int idx, int arrayIndex, int elementIndex, int remaining) { + private void arrayCopy( + Object[] value, + BitMap[] bitMaps, + TSStatus[] results, + int idx, + int arrayIndex, + int elementIndex, + int remaining) { for (int i = 0; i < values.size(); i++) { - if (value[i] == null) { + if (value[i] == null || !containsNonNullValue(bitMaps, results, i, idx, remaining)) { continue; } - List columnValues = values.get(i); + Object valueArray = getOrCreateValueArray(i, arrayIndex); switch (dataTypes.get(i)) { case TEXT: case BLOB: case STRING: case OBJECT: - Binary[] arrayT = ((Binary[]) columnValues.get(arrayIndex)); + Binary[] arrayT = (Binary[]) valueArray; System.arraycopy(value[i], idx, arrayT, elementIndex, remaining); // update raw size of Text chunk @@ -1028,25 +1034,25 @@ private void arrayCopy(Object[] value, int idx, int arrayIndex, int elementIndex } break; case FLOAT: - float[] arrayF = ((float[]) columnValues.get(arrayIndex)); + float[] arrayF = (float[]) valueArray; System.arraycopy(value[i], idx, arrayF, elementIndex, remaining); break; case INT32: case DATE: - int[] arrayI = ((int[]) columnValues.get(arrayIndex)); + int[] arrayI = (int[]) valueArray; System.arraycopy(value[i], idx, arrayI, elementIndex, remaining); break; case INT64: case TIMESTAMP: - long[] arrayL = ((long[]) columnValues.get(arrayIndex)); + long[] arrayL = (long[]) valueArray; System.arraycopy(value[i], idx, arrayL, elementIndex, remaining); break; case DOUBLE: - double[] arrayD = ((double[]) columnValues.get(arrayIndex)); + double[] arrayD = (double[]) valueArray; System.arraycopy(value[i], idx, arrayD, elementIndex, remaining); break; case BOOLEAN: - boolean[] arrayB = ((boolean[]) columnValues.get(arrayIndex)); + boolean[] arrayB = (boolean[]) valueArray; System.arraycopy(value[i], idx, arrayB, elementIndex, remaining); break; default: @@ -1055,6 +1061,39 @@ private void arrayCopy(Object[] value, int idx, int arrayIndex, int elementIndex } } + private static boolean containsNonNullValue( + BitMap[] bitMaps, TSStatus[] results, int columnIndex, int start, int length) { + BitMap bitMap = bitMaps == null ? null : bitMaps[columnIndex]; + boolean containsNull = bitMap != null && containsMarkedBit(bitMap, start, length); + boolean containsFailure = results != null && containsFailedStatus(results, start, length); + if (!containsNull && !containsFailure) { + return true; + } + for (int i = start; i < start + length; i++) { + boolean isNull = + (bitMap != null && bitMap.isMarked(i)) + || (results != null + && results[i] != null + && results[i].code != TSStatusCode.SUCCESS_STATUS.getStatusCode()); + if (!isNull) { + return true; + } + } + return false; + } + + private Object getOrCreateValueArray(int columnIndex, int arrayIndex) { + List columnValues = values.get(columnIndex); + Object valueArray = columnValues.get(arrayIndex); + if (valueArray == null) { + valueArray = getPrimitiveArraysByType(dataTypes.get(columnIndex)); + columnValues.set(arrayIndex, valueArray); + materializedValueArrayCounts[columnIndex]++; + materializedValueArrayMemCost += primitiveArrayMemCost(dataTypes.get(columnIndex)); + } + return valueArray; + } + private BitMap getBitMap(int columnIndex, int arrayIndex) { // init BitMaps if doesn't have if (bitMaps == null) { @@ -1101,7 +1140,12 @@ public TSDataType getDataType() { @Override public synchronized RamInfo calculateRamSize() { return new RamInfo( - timestamps.size(), alignedTvListArrayMemCost(), rowCount, new ArrayList<>(dataTypes)); + timestamps.size(), + alignedTvListArrayMemCost(), + (long) timestamps.size() * alignedTvListArrayMemCostWithoutPrimitiveArrays() + + materializedValueArrayMemCost, + rowCount, + new ArrayList<>(dataTypes)); } /** @@ -1168,6 +1212,29 @@ public long alignedTvListArrayMemCost() { return size; } + public long alignedTvListArrayMemCostWithoutPrimitiveArrays() { + long size = alignedTvListArrayMemCost(); + for (TSDataType dataType : dataTypes) { + if (dataType != null) { + size -= primitiveArrayMemCost(dataType); + } + } + return size; + } + + public static long alignedTvListArrayMemCostWithoutPrimitiveArrays( + TSDataType[] types, TsTableColumnCategory[] columnCategories) { + long size = alignedTvListArrayMemCost(types, columnCategories); + for (int i = 0; i < types.length; i++) { + TSDataType dataType = types[i]; + if (dataType != null + && (columnCategories == null || columnCategories[i] == TsTableColumnCategory.FIELD)) { + size -= primitiveArrayMemCost(dataType); + } + } + return size; + } + /** * Get the single column array mem cost by give type. * @@ -1175,14 +1242,20 @@ public long alignedTvListArrayMemCost() { * @return valueListArrayMemCost */ public static long valueListArrayMemCost(TSDataType type) { + return primitiveArrayMemCost(type) + valueListArrayMemCostWithoutPrimitiveArray(); + } + + public static long primitiveArrayMemCost(TSDataType type) { + // value array payload and header + return (long) PrimitiveArrayManager.ARRAY_SIZE * (long) type.getDataTypeSize() + + NUM_BYTES_ARRAY_HEADER; + } + + public static long valueListArrayMemCostWithoutPrimitiveArray() { long size = 0; - // value array mem size - size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long) type.getDataTypeSize(); // bitmap object, byte array, and reference in the bitmap list size += BITMAP_RAM_COST_PER_BLOCK; - // array headers mem size - size += NUM_BYTES_ARRAY_HEADER; - // Object references size in ArrayList + // null placeholder reference in the value ArrayList size += NUM_BYTES_OBJECT_REF; return size; } @@ -1404,7 +1477,13 @@ public int serializedSize() { case STRING: case OBJECT: for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) { - size += ReadWriteIOUtils.sizeToWrite(getBinaryByValueIndex(rowIdx, columnIndex)); + int arrayIndex = rowIdx / ARRAY_SIZE; + Object valueArray = values.get(columnIndex).get(arrayIndex); + Binary value = + valueArray == null + ? Binary.EMPTY_VALUE + : ((Binary[]) valueArray)[rowIdx % ARRAY_SIZE]; + size += ReadWriteIOUtils.sizeToWrite(value == null ? Binary.EMPTY_VALUE : value); } break; case FLOAT: @@ -1458,13 +1537,15 @@ public void serializeToWAL(IWALByteBufferView buffer) { for (int rowIndex = 0; rowIndex < rowCount; ++rowIndex) { int arrayIndex = rowIndex / ARRAY_SIZE; int elementIndex = rowIndex % ARRAY_SIZE; + Object valueArray = columnValues.get(arrayIndex); // value switch (dataTypes.get(columnIndex)) { case TEXT: case BLOB: case STRING: case OBJECT: - Binary valueT = ((Binary[]) columnValues.get(arrayIndex))[elementIndex]; + Binary valueT = + valueArray == null ? Binary.EMPTY_VALUE : ((Binary[]) valueArray)[elementIndex]; // In some scenario, the Binary in AlignedTVList will be null if this field is empty in // current row. We need to handle this scenario to get rid of NPE. See the similar issue // here: https://github.com/apache/iotdb/pull/9884 @@ -1474,29 +1555,29 @@ public void serializeToWAL(IWALByteBufferView buffer) { if (valueT != null) { WALWriteUtils.write(valueT, buffer); } else { - WALWriteUtils.write(new Binary(new byte[0]), buffer); + WALWriteUtils.write(Binary.EMPTY_VALUE, buffer); } break; case FLOAT: - float valueF = ((float[]) columnValues.get(arrayIndex))[elementIndex]; + float valueF = valueArray == null ? 0 : ((float[]) valueArray)[elementIndex]; buffer.putFloat(valueF); break; case INT32: case DATE: - int valueI = ((int[]) columnValues.get(arrayIndex))[elementIndex]; + int valueI = valueArray == null ? 0 : ((int[]) valueArray)[elementIndex]; buffer.putInt(valueI); break; case INT64: case TIMESTAMP: - long valueL = ((long[]) columnValues.get(arrayIndex))[elementIndex]; + long valueL = valueArray == null ? 0 : ((long[]) valueArray)[elementIndex]; buffer.putLong(valueL); break; case DOUBLE: - double valueD = ((double[]) columnValues.get(arrayIndex))[elementIndex]; + double valueD = valueArray == null ? 0 : ((double[]) valueArray)[elementIndex]; buffer.putDouble(valueD); break; case BOOLEAN: - boolean valueB = ((boolean[]) columnValues.get(arrayIndex))[elementIndex]; + boolean valueB = valueArray != null && ((boolean[]) valueArray)[elementIndex]; WALWriteUtils.write(valueB, buffer); break; default: diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java index f96473781aea0..3141ad3c8b442 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java @@ -67,19 +67,30 @@ public abstract class TVList implements WALEntryValue { public static class RamInfo { private final int timestampsSize; private final long arrayMemCost; + private final long ramSize; private final int rowCount; private final List dataTypes; public RamInfo( int timestampCount, long arrayMemCost, int rowCount, List dataTypes) { + this(timestampCount, arrayMemCost, (long) timestampCount * arrayMemCost, rowCount, dataTypes); + } + + public RamInfo( + int timestampCount, + long arrayMemCost, + long ramSize, + int rowCount, + List dataTypes) { this.timestampsSize = timestampCount; this.rowCount = rowCount; this.arrayMemCost = arrayMemCost; + this.ramSize = ramSize; this.dataTypes = dataTypes; } public long getRamSize() { - return timestampsSize * arrayMemCost; + return ramSize; } public int getTimestampsSize() { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index fbd2cfaf7bb44..3a85754a71c73 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -45,6 +45,7 @@ import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.utils.constant.TestConstant; +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -565,7 +566,7 @@ public void alignedTvListRamCostTest() } @Test - public void alignedTabletKeepsFailedStatusesAndCountsWrittenRows() + public void alignedTabletDoesNotChargePrimitiveArrayForFailedOnlyBlock() throws MetadataException, WriteProcessException, IOException, IllegalPathException { final int rowCount = PrimitiveArrayManager.ARRAY_SIZE + 2; final List rangeList = Collections.singletonList(new int[] {0, rowCount - 1}); @@ -585,12 +586,66 @@ public void alignedTabletKeepsFailedStatusesAndCountsWrittenRows() genSingleMeasurementTablet(rowCount, true), rangeList, actualResults, false, new long[5]); Assert.assertEquals( - expectedProcessor.getWorkMemTable().getTVListsRamCost(), + expectedProcessor.getWorkMemTable().getTVListsRamCost() + - AlignedTVList.primitiveArrayMemCost(dataType), actualProcessor.getWorkMemTable().getTVListsRamCost()); Assert.assertEquals( TSStatusCode.OUT_OF_TTL.getStatusCode(), actualResults[failedIndex].getCode()); } + @Test + public void alignedTabletOnlyChargesMaterializedPrimitiveArrays() + throws MetadataException, WriteProcessException, IOException, IllegalPathException { + processor = newTestProcessor(filePath + ".lazy-allocation"); + processor.insertTablet( + genAlignedTablet(new String[] {"s0", "s1"}, PrimitiveArrayManager.ARRAY_SIZE, 0), + Collections.singletonList(new int[] {0, PrimitiveArrayManager.ARRAY_SIZE}), + new TSStatus[PrimitiveArrayManager.ARRAY_SIZE], + true, + new long[5]); + + long ramCostBeforeNewBlock = processor.getWorkMemTable().getTVListsRamCost(); + processor.insertTablet( + genAlignedTablet(new String[] {"s0"}, 1, PrimitiveArrayManager.ARRAY_SIZE), + Collections.singletonList(new int[] {0, 1}), + new TSStatus[1], + true, + new long[5]); + + long denseBlockCost = + AlignedTVList.alignedTvListArrayMemCost( + new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null); + Assert.assertEquals( + denseBlockCost - AlignedTVList.primitiveArrayMemCost(TSDataType.INT32), + processor.getWorkMemTable().getTVListsRamCost() - ramCostBeforeNewBlock); + } + + @Test + public void alignedRowOnlyChargesMaterializedPrimitiveArrays() + throws MetadataException, WriteProcessException, IOException, IllegalPathException { + processor = newTestProcessor(filePath + ".lazy-row-allocation"); + processor.insertTablet( + genAlignedTablet(new String[] {"s0", "s1"}, PrimitiveArrayManager.ARRAY_SIZE, 0), + Collections.singletonList(new int[] {0, PrimitiveArrayManager.ARRAY_SIZE}), + new TSStatus[PrimitiveArrayManager.ARRAY_SIZE], + true, + new long[5]); + + long ramCostBeforeNewBlock = processor.getWorkMemTable().getTVListsRamCost(); + TSRecord record = new TSRecord(deviceId, PrimitiveArrayManager.ARRAY_SIZE); + record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, "s0", "1")); + InsertRowNode rowNode = buildInsertRowNodeByTSRecord(record); + rowNode.setAligned(true); + processor.insert(rowNode, new long[5]); + + long denseBlockCost = + AlignedTVList.alignedTvListArrayMemCost( + new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null); + Assert.assertEquals( + denseBlockCost - AlignedTVList.primitiveArrayMemCost(TSDataType.INT32), + processor.getWorkMemTable().getTVListsRamCost() - ramCostBeforeNewBlock); + } + @Test public void alignedTvListRamCostTest2() throws MetadataException, WriteProcessException, IOException { @@ -656,7 +711,7 @@ public void alignedTvListRamCostTest2() new TSStatus[10], true, new long[5]); - Assert.assertEquals(7009104, memTable.getTVListsRamCost()); + Assert.assertEquals(5425104, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(300, true), Collections.singletonList(new int[] {0, 10}), @@ -933,6 +988,31 @@ record = new TSRecord("root.vehicle.d2", 102); Assert.assertEquals(memTable1.memSize(), memTable2.memSize()); } + @Test + public void testAlignedSparseRowDoesNotChargeUnallocatedPrimitiveArrayOnNewBlock() + throws IOException, WriteProcessException, IllegalPathException { + processor = newTestProcessor(filePath); + String sparseDevice = deviceId + ".sparse"; + String denseDevice = deviceId + ".dense"; + for (int i = 0; i < PrimitiveArrayManager.ARRAY_SIZE; i++) { + insertAlignedRow(processor, sparseDevice, i, true); + insertAlignedRow(processor, denseDevice, i, true); + } + + IMemTable memTable = processor.getWorkMemTable(); + long ramCostBeforeSparseRow = memTable.getTVListsRamCost(); + insertAlignedRow(processor, sparseDevice, PrimitiveArrayManager.ARRAY_SIZE, false); + long sparseRowRamIncrement = memTable.getTVListsRamCost() - ramCostBeforeSparseRow; + + long ramCostBeforeDenseRow = memTable.getTVListsRamCost(); + insertAlignedRow(processor, denseDevice, PrimitiveArrayManager.ARRAY_SIZE, true); + long denseRowRamIncrement = memTable.getTVListsRamCost() - ramCostBeforeDenseRow; + + Assert.assertEquals( + AlignedTVList.primitiveArrayMemCost(dataType), + denseRowRamIncrement - sparseRowRamIncrement); + } + @Test public void testAlignedRamCostIgnoresRelationalNonFieldAndNullFieldColumns() throws IllegalPathException, WriteProcessException, IOException { @@ -1308,6 +1388,22 @@ private TsFileProcessor newTestProcessor(String path) throws IOException, WriteP return newProcessor; } + private void insertAlignedRow( + TsFileProcessor targetProcessor, + String targetDevice, + long timestamp, + boolean writeSecondColumn) + throws IllegalPathException, WriteProcessException { + TSRecord record = new TSRecord(targetDevice, timestamp); + record.addTuple(DataPoint.getDataPoint(dataType, measurementId, Long.toString(timestamp))); + if (writeSecondColumn) { + record.addTuple(DataPoint.getDataPoint(dataType, "s1", Long.toString(timestamp))); + } + InsertRowNode node = buildInsertRowNodeByTSRecord(record); + node.setAligned(true); + targetProcessor.insert(node, new long[5]); + } + private InsertTabletNode genSingleMeasurementTablet(int rowCount, boolean isAligned) throws IllegalPathException { String[] measurements = new String[] {measurementId}; @@ -1335,6 +1431,36 @@ private InsertTabletNode genSingleMeasurementTablet(int rowCount, boolean isAlig rowCount); } + private InsertTabletNode genAlignedTablet(String[] measurements, int rowCount, long startTime) + throws IllegalPathException { + TSDataType[] dataTypes = new TSDataType[measurements.length]; + MeasurementSchema[] schemas = new MeasurementSchema[measurements.length]; + Object[] columns = new Object[measurements.length]; + for (int i = 0; i < measurements.length; i++) { + dataTypes[i] = TSDataType.INT32; + schemas[i] = new MeasurementSchema(measurements[i], TSDataType.INT32, encoding); + columns[i] = new int[rowCount]; + } + long[] times = new long[rowCount]; + for (int row = 0; row < rowCount; row++) { + times[row] = startTime + row; + for (Object column : columns) { + ((int[]) column)[row] = row; + } + } + return new InsertTabletNode( + new QueryId("test_write").genPlanNodeId(), + new PartialPath(deviceId), + true, + measurements, + dataTypes, + schemas, + times, + null, + columns, + rowCount); + } + private InsertTabletNode genInsertTableNode(long startTime, boolean isAligned) throws IllegalPathException { String deviceId = "root.sg.device5"; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java index c6ccd333942ab..0b6feb7834951 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.utils.datastructure; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.common.conf.TSFileConfig; @@ -29,6 +30,10 @@ import org.junit.Assert; import org.junit.Test; +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -224,6 +229,144 @@ public void testEmptyInputBitmapsDoNotMaterializeMemTableBitmaps() { Assert.assertNull(tvList.getBitMaps()); } + @Test + public void testPrimitiveArraysAreAllocatedOnFirstWrite() { + AlignedTVList tvList = + AlignedTVList.newAlignedList( + new ArrayList<>(Arrays.asList(TSDataType.INT64, TSDataType.INT64))); + for (int i = 0; i <= ARRAY_SIZE; i++) { + tvList.putAlignedValue(i, new Object[] {(long) i, null}); + } + + Assert.assertNotNull(tvList.getValues().get(0).get(0)); + Assert.assertNotNull(tvList.getValues().get(0).get(1)); + Assert.assertNull(tvList.getValues().get(1).get(0)); + Assert.assertNull(tvList.getValues().get(1).get(1)); + + tvList.putAlignedValue(ARRAY_SIZE + 1L, new Object[] {null, 1L}); + + Assert.assertNull(tvList.getValues().get(1).get(0)); + Assert.assertNotNull(tvList.getValues().get(1).get(1)); + Assert.assertTrue(tvList.isNullValue(0, 1)); + Assert.assertEquals(1, tvList.getLongByValueIndex(ARRAY_SIZE + 1, 1)); + + tvList.extendColumn(TSDataType.INT32); + + Assert.assertNull(tvList.getValues().get(2).get(0)); + Assert.assertNull(tvList.getValues().get(2).get(1)); + + long ramSizeBeforeExtendedColumnMaterialization = tvList.calculateRamSize().getRamSize(); + tvList.putAlignedValue(ARRAY_SIZE + 2L, new Object[] {null, null, 2}); + + Assert.assertNull(tvList.getValues().get(2).get(0)); + Assert.assertNotNull(tvList.getValues().get(2).get(1)); + Assert.assertTrue(tvList.isNullValue(0, 2)); + Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE + 2, 2)); + Assert.assertEquals(2, tvList.getIntByValueIndex(ARRAY_SIZE + 2, 2)); + Assert.assertEquals( + AlignedTVList.primitiveArrayMemCost(TSDataType.INT32), + tvList.calculateRamSize().getRamSize() - ramSizeBeforeExtendedColumnMaterialization); + } + + @Test + public void testCalculateRamSizeCountsMaterializedPrimitiveArrays() { + AlignedTVList tvList = + AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64, TSDataType.INT64)); + for (int i = 0; i <= ARRAY_SIZE; i++) { + tvList.putAlignedValue(i, new Object[] {(long) i, null}); + } + + long ramSizeBeforeMaterialization = tvList.calculateRamSize().getRamSize(); + tvList.putAlignedValue(ARRAY_SIZE + 1L, new Object[] {1L, 1L}); + + Assert.assertEquals( + AlignedTVList.primitiveArrayMemCost(TSDataType.INT64), + tvList.calculateRamSize().getRamSize() - ramSizeBeforeMaterialization); + + Assert.assertEquals( + tvList.calculateRamSize().getRamSize(), tvList.clone().calculateRamSize().getRamSize()); + Assert.assertEquals( + tvList.calculateRamSize().getRamSize(), + tvList.cloneForFlushSort().calculateRamSize().getRamSize()); + + AlignedTVList projectedTvList = + (AlignedTVList) tvList.getTvListByColumnIndex(List.of(1), List.of(TSDataType.INT64), false); + Assert.assertEquals( + (long) projectedTvList.getValues().get(0).size() + * projectedTvList.alignedTvListArrayMemCostWithoutPrimitiveArrays() + + AlignedTVList.primitiveArrayMemCost(TSDataType.INT64), + projectedTvList.calculateRamSize().getRamSize()); + + tvList.clear(); + Assert.assertEquals(0, tvList.calculateRamSize().getRamSize()); + } + + @Test + public void testCalculateRamSizeExcludesUnallocatedPrimitiveArrays() { + AlignedTVList tvList = + AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64, TSDataType.INT64)); + for (int i = 0; i <= ARRAY_SIZE; i++) { + tvList.putAlignedValue(i, new Object[] {(long) i, null}); + } + + int blockCount = tvList.getValues().get(0).size(); + long denseRamSize = blockCount * tvList.alignedTvListArrayMemCost(); + long expectedRamSize = + denseRamSize - blockCount * AlignedTVList.primitiveArrayMemCost(TSDataType.INT64); + + Assert.assertEquals(expectedRamSize, tvList.calculateRamSize().getRamSize()); + } + + @Test + public void testBatchDoesNotAllocateAllNullPrimitiveArray() { + AlignedTVList tvList = + AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64, TSDataType.INT64)); + long[] times = new long[ARRAY_SIZE]; + long[][] values = new long[2][ARRAY_SIZE]; + BitMap[] bitMaps = new BitMap[] {null, new BitMap(ARRAY_SIZE)}; + bitMaps[1].markAll(); + for (int i = 0; i < ARRAY_SIZE; i++) { + times[i] = i; + values[0][i] = i; + values[1][i] = i; + } + + tvList.putAlignedValues(times, values, bitMaps, 0, ARRAY_SIZE, null); + + Assert.assertNotNull(tvList.getValues().get(0).get(0)); + Assert.assertNull(tvList.getValues().get(1).get(0)); + Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE - 1, 1)); + } + + @Test + public void testNullPrimitiveArrayCanBeClonedAndSerialized() throws IOException { + AlignedTVList tvList = + AlignedTVList.newAlignedList(Arrays.asList(TSDataType.TEXT, TSDataType.INT64)); + for (int i = 0; i <= ARRAY_SIZE; i++) { + tvList.putAlignedValue(i, new Object[] {null, null}); + } + tvList.putAlignedValue( + ARRAY_SIZE + 1L, new Object[] {new Binary("value", TSFileConfig.STRING_CHARSET), 1L}); + + AlignedTVList clonedTvList = tvList.clone(); + Assert.assertNull(clonedTvList.getValues().get(0).get(0)); + Assert.assertNull(clonedTvList.getValues().get(1).get(0)); + Assert.assertEquals("[null, null]", clonedTvList.getAlignedValue(0).toString()); + Assert.assertEquals("[value, 1]", clonedTvList.getAlignedValue(ARRAY_SIZE + 1).toString()); + + WALByteBufferForTest walBuffer = + new WALByteBufferForTest(ByteBuffer.allocate(tvList.serializedSize())); + tvList.serializeToWAL(walBuffer); + AlignedTVList deserializedTvList = + AlignedTVList.deserialize( + new DataInputStream(new ByteArrayInputStream(walBuffer.getBuffer().array()))); + + Assert.assertEquals(tvList.rowCount(), deserializedTvList.rowCount()); + Assert.assertEquals("[null, null]", deserializedTvList.getAlignedValue(0).toString()); + Assert.assertEquals( + "[value, 1]", deserializedTvList.getAlignedValue(ARRAY_SIZE + 1).toString()); + } + @Test public void testClone() { List dataTypes = new ArrayList<>();