This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 02b1ecd3775 [Performance] Allocate AlignedTVList primitive arrays
lazily (#18276)
02b1ecd3775 is described below
commit 02b1ecd37759c9211bcfe1ef7bc365db18967f9b
Author: Jiang Tian <[email protected]>
AuthorDate: Thu Jul 23 09:38:10 2026 +0800
[Performance] Allocate AlignedTVList primitive arrays lazily (#18276)
* Initialize primitive array lazily
* add perf test
* Add tests for lazy aligned array accounting
* fix memory calculation
---------
Co-authored-by: Caideyipi <[email protected]>
---
.../db/it/IoTDBAlignedTVListLazyAllocationIT.java | 142 +++++
...DBAlignedTVListPrimitiveArrayPerformanceIT.java | 584 +++++++++++++++++++++
.../memtable/AlignedWritableMemChunk.java | 4 +
.../dataregion/memtable/TsFileProcessor.java | 295 +++++++++--
.../db/utils/datastructure/AlignedTVList.java | 229 +++++---
.../iotdb/db/utils/datastructure/TVList.java | 13 +-
.../dataregion/memtable/TsFileProcessorTest.java | 132 ++++-
.../db/utils/datastructure/AlignedTVListTest.java | 143 +++++
8 files changed, 1407 insertions(+), 135 deletions(-)
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 00000000000..375f560c9ee
--- /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<String> measurements = new ArrayList<>(NEW_COLUMN_COUNT);
+ List<TSDataType> dataTypes = new ArrayList<>(NEW_COLUMN_COUNT);
+ List<Object> 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<org.apache.tsfile.read.common.Field> 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<IMeasurementSchema> 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 00000000000..c0bbf499062
--- /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<String> newMeasurements = new ArrayList<>(NEW_COLUMN_COUNT);
+ List<TSDataType> newDataTypes = new ArrayList<>(NEW_COLUMN_COUNT);
+ List<Object> 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<String> newMeasurements,
+ List<TSDataType> newDataTypes,
+ List<Object> 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<IMeasurementSchema> 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<IMeasurementSchema> 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<org.apache.tsfile.read.common.Field> 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<org.apache.tsfile.read.common.Field> 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-worktree> 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=<master-commit>`, and `-D%s=<baseline-result>`.%n"
+ + "5. Run the candidate with the corresponding candidate
values plus "
+ + "`-D%s=<baseline-result>` and `-D%s=<report-path>`. 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 b0d00562a4c..05fd1d4073c 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 class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
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 4528b44f598..8b14459a760 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.IDeviceID;
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.ArrayList;
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 @@ public class TsFileProcessor {
insertTabletNode.getMeasurements(),
insertTabletNode.getDataTypes(),
insertTabletNode.getColumns(),
+ insertTabletNode.getBitMaps(),
insertTabletNode.getColumnCategories(),
splitStart,
splitEnd,
@@ -714,9 +718,11 @@ public class TsFileProcessor {
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 @@ public class TsFileProcessor {
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 @@ public class TsFileProcessor {
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<TSDataType> 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 @@ public class TsFileProcessor {
@SuppressWarnings("squid:S3776") // high Cognitive Complexity
private long[] checkAlignedMemCostAndAddToTspInfoForRows(List<InsertRowNode>
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<IDeviceID, Pair<Map<String, TSDataType>, Integer>>
increasingMemTableInfo = new HashMap<>();
+ Map<IDeviceID, Map<String, Set<Integer>>> materializedArraysInCurrentBatch
= new HashMap<>();
for (InsertRowNode insertRowNode : insertRowNodeList) {
IDeviceID deviceId = insertRowNode.getDeviceID();
TSDataType[] dataTypes = insertRowNode.getDataTypes();
@@ -888,11 +914,13 @@ public class TsFileProcessor {
measurements, dataTypes, values,
insertRowNode.getColumnCategories());
Pair<Map<String, TSDataType>, 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 @@ public class TsFileProcessor {
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 @@ public class TsFileProcessor {
// For existed device of this mem table
AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk)
memChunk;
int currentChunkPointNum = alignedMemChunk == null ? 0 :
alignedMemChunk.alignedListSize();
- List<TSDataType> dataTypesInTVList = new ArrayList<>();
Pair<Map<String, TSDataType>, 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 @@ public class TsFileProcessor {
continue;
}
- int addingPointNum = addingPointNumInfo.getRight();
// Extending the column of aligned mem chunk
boolean currentMemChunkContainsMeasurement =
alignedMemChunk != null &&
alignedMemChunk.containsMeasurement(measurements[i]);
@@ -926,27 +959,46 @@ public class TsFileProcessor {
&& !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<Integer> 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 @@ public class TsFileProcessor {
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 @@ public class TsFileProcessor {
String[] measurements,
TSDataType[] dataTypes,
Object[] columns,
+ BitMap[] bitMaps,
TsTableColumnCategory[] columnCategories,
int start,
int end,
@@ -1018,6 +1071,7 @@ public class TsFileProcessor {
end,
memIncrements,
columns,
+ bitMaps,
columnCategories,
results);
long memTableIncrement = memIncrements[0];
@@ -1035,7 +1089,9 @@ public class TsFileProcessor {
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 @@ public class TsFileProcessor {
int end,
long[] memIncrements,
Object[] columns,
+ BitMap[] bitMaps,
TsTableColumnCategory[] columnCategories,
TSStatus[] results) {
int incomingPointNum = end - start;
@@ -1085,21 +1142,38 @@ public class TsFileProcessor {
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<TSDataType> dataTypesInTVList = new ArrayList<>();
+ List<TSDataType> newDataTypes = new ArrayList<>();
int currentPointNum = alignedMemChunk.alignedListSize();
int newPointNum = currentPointNum + incomingPointNum;
int currentArrayCnt =
@@ -1112,9 +1186,12 @@ public class TsFileProcessor {
}
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 @@ public class TsFileProcessor {
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 @@ public class TsFileProcessor {
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<Object> 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 464cbe09995..644890d0bed 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<List<Object>> 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 @@ public abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
}
}
cloneList.timeColDeletedMap = timeColDeletedMap == null ? null :
timeColDeletedMap.clone();
+ cloneList.materializedValueArrayCounts =
+ Arrays.copyOf(materializedValueArrayCounts,
materializedValueArrayCounts.length);
+ cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost;
return cloneList;
}
@@ -229,43 +251,35 @@ public abstract class AlignedTVList extends TVList {
timestamps.get(arrayIndex)[elementIndex] = timestamp;
for (int i = 0; i < values.size(); i++) {
Object columnValue = value[i];
- List<Object> 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 abstract class AlignedTVList extends TVList {
List<Object> columnValue = new ArrayList<>(timestamps.size());
List<BitMap> 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 abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
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<BitMap> columnBitMaps = bitMaps.get(columnIndex);
return columnBitMaps.get(arrayIndex).isMarked(elementIndex);
@@ -744,6 +736,9 @@ public abstract class AlignedTVList extends TVList {
}
protected Object cloneValue(TSDataType type, Object value) {
+ if (value == null) {
+ return null;
+ }
switch (type) {
case TEXT:
case BLOB:
@@ -791,12 +786,16 @@ public abstract class AlignedTVList extends TVList {
List<Object> 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 @@ public abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
// 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 @@ public abstract class AlignedTVList extends TVList {
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<Object> 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 @@ public abstract class AlignedTVList extends TVList {
}
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 @@ public abstract class AlignedTVList extends TVList {
}
}
+ 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<Object> 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 abstract class AlignedTVList extends TVList {
@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 abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
* @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 abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
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 abstract class AlignedTVList extends TVList {
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 f96473781ae..3141ad3c8b4 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<TSDataType> dataTypes;
public RamInfo(
int timestampCount, long arrayMemCost, int rowCount, List<TSDataType>
dataTypes) {
+ this(timestampCount, arrayMemCost, (long) timestampCount * arrayMemCost,
rowCount, dataTypes);
+ }
+
+ public RamInfo(
+ int timestampCount,
+ long arrayMemCost,
+ long ramSize,
+ int rowCount,
+ List<TSDataType> 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 fbd2cfaf7bb..3a85754a71c 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.PrimitiveArrayManager;
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 class TsFileProcessorTest {
}
@Test
- public void alignedTabletKeepsFailedStatusesAndCountsWrittenRows()
+ public void alignedTabletDoesNotChargePrimitiveArrayForFailedOnlyBlock()
throws MetadataException, WriteProcessException, IOException,
IllegalPathException {
final int rowCount = PrimitiveArrayManager.ARRAY_SIZE + 2;
final List<int[]> rangeList = Collections.singletonList(new int[] {0,
rowCount - 1});
@@ -585,12 +586,66 @@ public class TsFileProcessorTest {
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 class TsFileProcessorTest {
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 @@ public class TsFileProcessorTest {
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 @@ public class TsFileProcessorTest {
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 @@ public class TsFileProcessorTest {
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 c6ccd333942..0b6feb78349 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.apache.tsfile.utils.BitMap;
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 class AlignedTVListTest {
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<TSDataType> dataTypes = new ArrayList<>();
