You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/04/21 12:55:45 UTC
[iotdb] 04/04: Add ColumnMerger to merge multipul input columns of same sensor into one column
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch TimeJoinOperatorBug
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 74c6de06fb94e6ef2a4ea8415570c1aac854ee82
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Thu Apr 21 20:55:23 2022 +0800
Add ColumnMerger to merge multipul input columns of same sensor into one column
---
.../db/mpp/operator/process/ColumnMerger.java | 27 -----
.../db/mpp/operator/process/TimeJoinOperator.java | 55 +++++-----
.../mpp/operator/process/merge/ColumnMerger.java | 58 ++++++++++
.../operator/process/merge/SingleColumnMerger.java | 74 +++++++++++++
.../db/mpp/sql/planner/LocalExecutionPlanner.java | 28 ++++-
.../db/mpp/sql/planner/plan/OutputColumn.java | 14 +++
.../iotdb/db/mpp/execution/DataDriverTest.java | 7 +-
.../iotdb/db/mpp/operator/LimitOperatorTest.java | 7 +-
.../db/mpp/operator/SingleColumnMergerTest.java | 117 ++++++++++++++++++++
.../db/mpp/operator/TimeJoinOperatorTest.java | 120 ++++++++++++++++++++-
.../common/block/column/BinaryColumnBuilder.java | 18 +---
.../common/block/column/BooleanColumnBuilder.java | 18 +---
.../read/common/block/column/ColumnBuilder.java | 17 ++-
.../common/block/column/DoubleColumnBuilder.java | 18 +---
.../common/block/column/FloatColumnBuilder.java | 14 +--
.../read/common/block/column/IntColumnBuilder.java | 18 +---
.../common/block/column/LongColumnBuilder.java | 18 +---
.../common/block/column/TimeColumnBuilder.java | 16 +--
18 files changed, 499 insertions(+), 145 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/ColumnMerger.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/ColumnMerger.java
deleted file mode 100644
index 19503b823d..0000000000
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/ColumnMerger.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * 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.mpp.operator.process;
-
-import org.apache.iotdb.tsfile.read.common.block.TsBlock;
-import org.apache.iotdb.tsfile.read.common.block.column.Column;
-
-public interface ColumnMerger {
-
- Column mergeColumn(TsBlock[] inputTsBlocks, int[] inputIndex, int[] updatedInputIndex);
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/TimeJoinOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/TimeJoinOperator.java
index 1b9cadeb39..0c0fbc96e1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/TimeJoinOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/TimeJoinOperator.java
@@ -20,20 +20,20 @@ package org.apache.iotdb.db.mpp.operator.process;
import org.apache.iotdb.db.mpp.operator.Operator;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.db.mpp.operator.process.merge.ColumnMerger;
import org.apache.iotdb.db.mpp.sql.statement.component.OrderBy;
import org.apache.iotdb.db.utils.datastructure.TimeSelector;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.iotdb.tsfile.read.common.block.column.Column;
-import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
-import org.apache.iotdb.tsfile.read.common.block.column.TimeColumn;
import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
import com.google.common.util.concurrent.ListenableFuture;
import java.util.List;
+import static com.google.common.base.Preconditions.checkArgument;
+
public class TimeJoinOperator implements ProcessOperator {
private final OperatorContext operatorContext;
@@ -44,8 +44,12 @@ public class TimeJoinOperator implements ProcessOperator {
private final TsBlock[] inputTsBlocks;
+ /** start index for each input TsBlocks and size of it is equal to inputTsBlocks */
private final int[] inputIndex;
+ /** used to record current index for input TsBlocks after merging */
+ private final int[] shadowInputIndex;
+
private final boolean[] noMoreTsBlocks;
private final TimeSelector timeSelector;
@@ -58,7 +62,9 @@ public class TimeJoinOperator implements ProcessOperator {
*/
private final List<TSDataType> dataTypes;
- private List<ColumnMerger> mergers;
+ private final List<ColumnMerger> mergers;
+
+ private final TsBlockBuilder tsBlockBuilder;
private boolean finished;
@@ -66,16 +72,23 @@ public class TimeJoinOperator implements ProcessOperator {
OperatorContext operatorContext,
List<Operator> children,
OrderBy mergeOrder,
- List<TSDataType> dataTypes) {
+ List<TSDataType> dataTypes,
+ List<ColumnMerger> mergers) {
+ checkArgument(
+ children != null && children.size() > 0,
+ "child size of TimeJoinOperator should be larger than 0");
this.operatorContext = operatorContext;
this.children = children;
this.inputCount = children.size();
this.inputTsBlocks = new TsBlock[this.inputCount];
this.inputIndex = new int[this.inputCount];
+ this.shadowInputIndex = new int[this.inputCount];
this.noMoreTsBlocks = new boolean[this.inputCount];
this.timeSelector = new TimeSelector(this.inputCount << 1, OrderBy.TIMESTAMP_ASC == mergeOrder);
this.columnCount = dataTypes.size();
this.dataTypes = dataTypes;
+ this.tsBlockBuilder = new TsBlockBuilder(dataTypes);
+ this.mergers = mergers;
}
@Override
@@ -98,12 +111,13 @@ public class TimeJoinOperator implements ProcessOperator {
@Override
public TsBlock next() {
+ tsBlockBuilder.reset();
// end time for returned TsBlock this time, it's the min end time among all the children
// TsBlocks
long currentEndTime = 0;
boolean init = false;
for (int i = 0; i < inputCount; i++) {
- if (!noMoreTsBlocks[i] && empty(i)) {
+ if (!noMoreTsBlocks[i] && empty(i) && children.get(i).hasNext()) {
inputIndex[i] = 0;
inputTsBlocks[i] = children.get(i).next();
if (!empty(i)) {
@@ -129,34 +143,26 @@ public class TimeJoinOperator implements ProcessOperator {
return tsBlockBuilder.build();
}
- TsBlockBuilder tsBlockBuilder = TsBlockBuilder.createWithOnlyTimeColumn();
-
TimeColumnBuilder timeBuilder = tsBlockBuilder.getTimeColumnBuilder();
while (!timeSelector.isEmpty() && timeSelector.first() <= currentEndTime) {
timeBuilder.writeLong(timeSelector.pollFirst());
tsBlockBuilder.declarePosition();
}
- tsBlockBuilder.buildValueColumnBuilders(dataTypes);
-
for (int i = 0; i < columnCount; i++) {
ColumnMerger merger = mergers.get(i);
-
+ merger.mergeColumn(
+ inputTsBlocks,
+ inputIndex,
+ shadowInputIndex,
+ timeBuilder,
+ currentEndTime,
+ tsBlockBuilder.getColumnBuilder(i));
}
- for (int i = 0, column = 0; i < inputCount; i++) {
- TsBlock block = inputTsBlocks[i];
- TimeColumn timeColumn = block.getTimeColumn();
- int valueColumnCount = block.getValueColumnCount();
- int startIndex = inputIndex[i];
- for (int j = 0; j < valueColumnCount; j++) {
- startIndex = inputIndex[i];
- ColumnBuilder columnBuilder = tsBlockBuilder.getColumnBuilder(column++);
- Column valueColumn = block.getColumn(j);
- startIndex = columnBuilder.appendColumn(timeColumn, valueColumn, startIndex, timeBuilder);
- }
- inputIndex[i] = startIndex;
- }
+ // update inputIndex using shadowInputIndex
+ System.arraycopy(shadowInputIndex, 0, inputIndex, 0, inputCount);
+
return tsBlockBuilder.build();
}
@@ -173,6 +179,7 @@ public class TimeJoinOperator implements ProcessOperator {
return true;
} else {
noMoreTsBlocks[i] = true;
+ inputTsBlocks[i] = null;
}
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/merge/ColumnMerger.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/merge/ColumnMerger.java
new file mode 100644
index 0000000000..04ed2bb09e
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/merge/ColumnMerger.java
@@ -0,0 +1,58 @@
+/*
+ * 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.mpp.operator.process.merge;
+
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
+
+/** used to merge columns belonging to same series into one column */
+public interface ColumnMerger {
+
+ /**
+ * @param tsBlockIndex index
+ * @param inputTsBlocks input TsBlock array
+ * @param inputIndex current index for each input TsBlock and size of it is equal to inputTsBlocks
+ * @return true if TsBlock at tsBlockIndex is null or its current read index is larger than its
+ * size
+ */
+ default boolean empty(int tsBlockIndex, TsBlock[] inputTsBlocks, int[] inputIndex) {
+ return inputTsBlocks[tsBlockIndex] == null
+ || inputTsBlocks[tsBlockIndex].getPositionCount() == inputIndex[tsBlockIndex];
+ }
+
+ /**
+ * merge columns belonging to same series into one column
+ *
+ * @param inputTsBlocks all source TsBlocks, some of which will cantain source column
+ * @param inputIndex start index for each source TsBlock and size of it is equal to inputTsBlocks
+ * @param updatedInputIndex current index for each source TsBlock after merging
+ * @param timeBuilder result time column, which is already generated and used to indicate each
+ * row's timestamp
+ * @param currentEndTime merge until each input column's time is larger than currentEndTime
+ * @param columnBuilder used to write merged value into
+ */
+ void mergeColumn(
+ TsBlock[] inputTsBlocks,
+ int[] inputIndex,
+ int[] updatedInputIndex,
+ TimeColumnBuilder timeBuilder,
+ long currentEndTime,
+ ColumnBuilder columnBuilder);
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/merge/SingleColumnMerger.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/merge/SingleColumnMerger.java
new file mode 100644
index 0000000000..e4e054a0ef
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/merge/SingleColumnMerger.java
@@ -0,0 +1,74 @@
+/*
+ * 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.mpp.operator.process.merge;
+
+import org.apache.iotdb.db.mpp.sql.planner.plan.InputLocation;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.column.Column;
+import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
+
+/** only has one input column */
+public class SingleColumnMerger implements ColumnMerger {
+
+ private final InputLocation location;
+
+ public SingleColumnMerger(InputLocation location) {
+ this.location = location;
+ }
+
+ public void mergeColumn(
+ TsBlock[] inputTsBlocks,
+ int[] inputIndex,
+ int[] updatedInputIndex,
+ TimeColumnBuilder timeBuilder,
+ long currentEndTime,
+ ColumnBuilder columnBuilder) {
+ int tsBlockIndex = location.getTsBlockIndex();
+ int columnIndex = location.getValueColumnIndex();
+
+ int rowCount = timeBuilder.getPositionCount();
+ int index = inputIndex[tsBlockIndex];
+ // input column is empty or current time of input column is already larger than currentEndTime
+ // just appendNull rowCount null
+ if (empty(tsBlockIndex, inputTsBlocks, inputIndex)
+ || inputTsBlocks[tsBlockIndex].getTimeByIndex(index) > currentEndTime) {
+ columnBuilder.appendNull(rowCount);
+ } else {
+ // read from input column and write it into columnBuilder
+ TimeColumn timeColumn = inputTsBlocks[tsBlockIndex].getTimeColumn();
+ Column valueColumn = inputTsBlocks[tsBlockIndex].getColumn(columnIndex);
+ for (int i = 0; i < rowCount; i++) {
+ // current index is less than size of input column and current time of input column is equal
+ // to result row's time and input column's value at index is not null
+ if (timeColumn.getPositionCount() > index
+ && timeColumn.getLong(index) == timeBuilder.getTime(i)
+ && !valueColumn.isNull(index)) {
+ columnBuilder.write(valueColumn, index++);
+ } else {
+ // otherwise, append a null
+ columnBuilder.appendNull();
+ }
+ }
+ // update the index after merging
+ updatedInputIndex[tsBlockIndex] = index;
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java
index e5d60439bd..79c22c1c6a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java
@@ -37,6 +37,8 @@ import org.apache.iotdb.db.mpp.operator.Operator;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
import org.apache.iotdb.db.mpp.operator.process.LimitOperator;
import org.apache.iotdb.db.mpp.operator.process.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.operator.process.merge.ColumnMerger;
+import org.apache.iotdb.db.mpp.operator.process.merge.SingleColumnMerger;
import org.apache.iotdb.db.mpp.operator.schema.DevicesSchemaScanOperator;
import org.apache.iotdb.db.mpp.operator.schema.SchemaFetchOperator;
import org.apache.iotdb.db.mpp.operator.schema.SchemaMergeOperator;
@@ -45,6 +47,7 @@ import org.apache.iotdb.db.mpp.operator.source.DataSourceOperator;
import org.apache.iotdb.db.mpp.operator.source.ExchangeOperator;
import org.apache.iotdb.db.mpp.operator.source.SeriesAggregateScanOperator;
import org.apache.iotdb.db.mpp.operator.source.SeriesScanOperator;
+import org.apache.iotdb.db.mpp.sql.planner.plan.OutputColumn;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.DevicesSchemaScanNode;
@@ -308,8 +311,31 @@ public class LocalExecutionPlanner {
context.getNextOperatorId(),
node.getPlanNodeId(),
TimeJoinOperator.class.getSimpleName());
+ List<OutputColumn> outputColumns = node.getOutputColumns();
+ List<ColumnMerger> mergers = createColumnMergers(outputColumns);
return new TimeJoinOperator(
- operatorContext, children, node.getMergeOrder(), node.getOutputColumnTypes());
+ operatorContext, children, node.getMergeOrder(), node.getOutputColumnTypes(), mergers);
+ }
+
+ private List<ColumnMerger> createColumnMergers(List<OutputColumn> outputColumns) {
+ List<ColumnMerger> mergers = new ArrayList<>(outputColumns.size());
+ for (OutputColumn outputColumn : outputColumns) {
+ ColumnMerger merger;
+ // only has one input column
+ if (outputColumn.isSingleInputColumn()) {
+ merger = new SingleColumnMerger(outputColumn.getInputLocation(0));
+ } else if (!outputColumn.isOverlapped()) {
+ // has more than one input columns but time of these input columns is not overlapped
+ throw new UnsupportedOperationException(
+ "has more than one input columns but time of these input columns is not overlapped is not supported");
+ } else {
+ // has more than one input columns and time of these input columns is overlapped
+ throw new UnsupportedOperationException(
+ "has more than one input columns and time of these input columns is overlapped is not supported");
+ }
+ mergers.add(merger);
+ }
+ return mergers;
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/OutputColumn.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/OutputColumn.java
index 2596619166..168027ef4f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/OutputColumn.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/OutputColumn.java
@@ -22,6 +22,8 @@ import com.google.common.collect.ImmutableList;
import java.util.List;
+import static com.google.common.base.Preconditions.checkArgument;
+
public class OutputColumn {
// indicate this output column should use which value column of which input TsBlock
@@ -41,6 +43,9 @@ public class OutputColumn {
}
public OutputColumn(List<InputLocation> sourceLocations, boolean overlapped) {
+ checkArgument(
+ sourceLocations != null && sourceLocations.size() > 0,
+ "size of sourceLocations should be larger than 0");
this.sourceLocations = sourceLocations;
this.overlapped = overlapped;
}
@@ -52,4 +57,13 @@ public class OutputColumn {
public boolean isOverlapped() {
return overlapped;
}
+
+ public boolean isSingleInputColumn() {
+ return sourceLocations.size() == 1;
+ }
+
+ public InputLocation getInputLocation(int index) {
+ checkArgument(index < sourceLocations.size(), "index is not valid");
+ return sourceLocations.get(index);
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
index 2703d0aa37..e69e6b6337 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
@@ -32,7 +32,9 @@ import org.apache.iotdb.db.mpp.common.PlanFragmentId;
import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.operator.process.LimitOperator;
import org.apache.iotdb.db.mpp.operator.process.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.operator.process.merge.SingleColumnMerger;
import org.apache.iotdb.db.mpp.operator.source.SeriesScanOperator;
+import org.apache.iotdb.db.mpp.sql.planner.plan.InputLocation;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.sql.statement.component.OrderBy;
import org.apache.iotdb.db.query.reader.series.SeriesReaderTestUtil;
@@ -136,7 +138,10 @@ public class DataDriverTest {
fragmentInstanceContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
OrderBy.TIMESTAMP_ASC,
- Arrays.asList(TSDataType.INT32, TSDataType.INT32));
+ Arrays.asList(TSDataType.INT32, TSDataType.INT32),
+ Arrays.asList(
+ new SingleColumnMerger(new InputLocation(0, 0)),
+ new SingleColumnMerger(new InputLocation(1, 0))));
LimitOperator limitOperator =
new LimitOperator(
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/operator/LimitOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/operator/LimitOperatorTest.java
index 6ad74bd110..5d83022379 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/operator/LimitOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/operator/LimitOperatorTest.java
@@ -30,7 +30,9 @@ import org.apache.iotdb.db.mpp.execution.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.FragmentInstanceState;
import org.apache.iotdb.db.mpp.operator.process.LimitOperator;
import org.apache.iotdb.db.mpp.operator.process.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.operator.process.merge.SingleColumnMerger;
import org.apache.iotdb.db.mpp.operator.source.SeriesScanOperator;
+import org.apache.iotdb.db.mpp.sql.planner.plan.InputLocation;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.sql.statement.component.OrderBy;
import org.apache.iotdb.db.query.reader.series.SeriesReaderTestUtil;
@@ -131,7 +133,10 @@ public class LimitOperatorTest {
fragmentInstanceContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
OrderBy.TIMESTAMP_ASC,
- Arrays.asList(TSDataType.INT32, TSDataType.INT32));
+ Arrays.asList(TSDataType.INT32, TSDataType.INT32),
+ Arrays.asList(
+ new SingleColumnMerger(new InputLocation(0, 0)),
+ new SingleColumnMerger(new InputLocation(1, 0))));
LimitOperator limitOperator =
new LimitOperator(
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/operator/SingleColumnMergerTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/operator/SingleColumnMergerTest.java
new file mode 100644
index 0000000000..706ba38a93
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/operator/SingleColumnMergerTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.mpp.operator;
+
+import org.apache.iotdb.db.mpp.operator.process.merge.SingleColumnMerger;
+import org.apache.iotdb.db.mpp.sql.planner.plan.InputLocation;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.Column;
+import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
+
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class SingleColumnMergerTest {
+
+ @Test
+ public void mergeTest1() {
+ SingleColumnMerger merger = new SingleColumnMerger(new InputLocation(0, 0));
+
+ TsBlockBuilder inputBuilder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+ inputBuilder.getTimeColumnBuilder().writeLong(2);
+ inputBuilder.getColumnBuilder(0).writeInt(20);
+ inputBuilder.declarePosition();
+ inputBuilder.getTimeColumnBuilder().writeLong(4);
+ inputBuilder.getColumnBuilder(0).writeInt(40);
+ inputBuilder.declarePosition();
+ inputBuilder.getTimeColumnBuilder().writeLong(6);
+ inputBuilder.getColumnBuilder(0).writeInt(60);
+ inputBuilder.declarePosition();
+
+ TsBlock[] inputTsBlocks = new TsBlock[] {inputBuilder.build()};
+ int[] inputIndex = new int[] {1};
+ int[] updatedInputIndex = new int[] {1};
+
+ TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+ TimeColumnBuilder timeColumnBuilder = builder.getTimeColumnBuilder();
+ timeColumnBuilder.writeLong(4);
+ builder.declarePosition();
+ timeColumnBuilder.writeLong(5);
+ builder.declarePosition();
+ timeColumnBuilder.writeLong(6);
+ builder.declarePosition();
+ timeColumnBuilder.writeLong(7);
+ builder.declarePosition();
+ ColumnBuilder valueColumnBuilder = builder.getColumnBuilder(0);
+
+ merger.mergeColumn(
+ inputTsBlocks, inputIndex, updatedInputIndex, timeColumnBuilder, 7, valueColumnBuilder);
+
+ Column result = valueColumnBuilder.build();
+
+ assertEquals(4, result.getPositionCount());
+ assertFalse(result.isNull(0));
+ assertEquals(40, result.getInt(0));
+ assertTrue(result.isNull(1));
+ assertFalse(result.isNull(2));
+ assertEquals(60, result.getInt(2));
+ assertTrue(result.isNull(3));
+ }
+
+ /** input tsblock is null */
+ @Test
+ public void mergeTest2() {
+ SingleColumnMerger merger = new SingleColumnMerger(new InputLocation(0, 0));
+
+ TsBlock[] inputTsBlocks = new TsBlock[1];
+ int[] inputIndex = new int[] {0};
+ int[] updatedInputIndex = new int[] {0};
+
+ TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+ TimeColumnBuilder timeColumnBuilder = builder.getTimeColumnBuilder();
+ timeColumnBuilder.writeLong(4);
+ builder.declarePosition();
+ timeColumnBuilder.writeLong(5);
+ builder.declarePosition();
+ timeColumnBuilder.writeLong(6);
+ builder.declarePosition();
+ timeColumnBuilder.writeLong(7);
+ builder.declarePosition();
+ ColumnBuilder valueColumnBuilder = builder.getColumnBuilder(0);
+
+ merger.mergeColumn(
+ inputTsBlocks, inputIndex, updatedInputIndex, timeColumnBuilder, 7, valueColumnBuilder);
+
+ Column result = valueColumnBuilder.build();
+
+ assertEquals(4, result.getPositionCount());
+ assertTrue(result.isNull(0));
+ assertTrue(result.isNull(1));
+ assertTrue(result.isNull(2));
+ assertTrue(result.isNull(3));
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/operator/TimeJoinOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/operator/TimeJoinOperatorTest.java
index 5534418b84..cbafb1a842 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/operator/TimeJoinOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/operator/TimeJoinOperatorTest.java
@@ -29,7 +29,9 @@ import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.execution.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.FragmentInstanceState;
import org.apache.iotdb.db.mpp.operator.process.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.operator.process.merge.SingleColumnMerger;
import org.apache.iotdb.db.mpp.operator.source.SeriesScanOperator;
+import org.apache.iotdb.db.mpp.sql.planner.plan.InputLocation;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.sql.statement.component.OrderBy;
import org.apache.iotdb.db.query.reader.series.SeriesReaderTestUtil;
@@ -37,6 +39,7 @@ import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.read.common.block.column.IntColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.RunLengthEncodedColumn;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
@@ -125,7 +128,10 @@ public class TimeJoinOperatorTest {
fragmentInstanceContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
OrderBy.TIMESTAMP_ASC,
- Arrays.asList(TSDataType.INT32, TSDataType.INT32));
+ Arrays.asList(TSDataType.INT32, TSDataType.INT32),
+ Arrays.asList(
+ new SingleColumnMerger(new InputLocation(0, 0)),
+ new SingleColumnMerger(new InputLocation(1, 0))));
int count = 0;
while (timeJoinOperator.hasNext()) {
TsBlock tsBlock = timeJoinOperator.next();
@@ -157,4 +163,116 @@ public class TimeJoinOperatorTest {
fail();
}
}
+
+ /** test time join with non-exist sensor */
+ @Test
+ public void batchTest2() {
+ try {
+ MeasurementPath measurementPath1 =
+ new MeasurementPath(TIME_JOIN_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
+ Set<String> allSensors = new HashSet<>();
+ allSensors.add("sensor0");
+ allSensors.add("sensor1");
+ allSensors.add("error_sensor");
+ QueryId queryId = new QueryId("stub_query");
+ AtomicReference<FragmentInstanceState> state =
+ new AtomicReference<>(FragmentInstanceState.RUNNING);
+ FragmentInstanceContext fragmentInstanceContext =
+ new FragmentInstanceContext(
+ new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance"), state);
+ PlanNodeId planNodeId1 = new PlanNodeId("1");
+ fragmentInstanceContext.addOperatorContext(
+ 1, planNodeId1, SeriesScanOperator.class.getSimpleName());
+ PlanNodeId planNodeId2 = new PlanNodeId("2");
+ fragmentInstanceContext.addOperatorContext(
+ 2, planNodeId2, SeriesScanOperator.class.getSimpleName());
+ PlanNodeId planNodeId3 = new PlanNodeId("3");
+ fragmentInstanceContext.addOperatorContext(
+ 3, planNodeId3, SeriesScanOperator.class.getSimpleName());
+ fragmentInstanceContext.addOperatorContext(
+ 4, new PlanNodeId("4"), TimeJoinOperator.class.getSimpleName());
+ SeriesScanOperator seriesScanOperator1 =
+ new SeriesScanOperator(
+ planNodeId1,
+ measurementPath1,
+ allSensors,
+ TSDataType.INT32,
+ fragmentInstanceContext.getOperatorContexts().get(0),
+ null,
+ null,
+ true);
+ seriesScanOperator1.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
+
+ MeasurementPath measurementPath2 =
+ new MeasurementPath(TIME_JOIN_OPERATOR_TEST_SG + ".device0.sensor1", TSDataType.INT32);
+ SeriesScanOperator seriesScanOperator2 =
+ new SeriesScanOperator(
+ planNodeId2,
+ measurementPath2,
+ allSensors,
+ TSDataType.INT32,
+ fragmentInstanceContext.getOperatorContexts().get(1),
+ null,
+ null,
+ true);
+ seriesScanOperator2.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
+
+ MeasurementPath measurementPath3 =
+ new MeasurementPath(
+ TIME_JOIN_OPERATOR_TEST_SG + ".device0.error_sensor", TSDataType.INT32);
+ SeriesScanOperator seriesScanOperator3 =
+ new SeriesScanOperator(
+ planNodeId3,
+ measurementPath3,
+ allSensors,
+ TSDataType.INT32,
+ fragmentInstanceContext.getOperatorContexts().get(2),
+ null,
+ null,
+ true);
+ seriesScanOperator3.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
+
+ TimeJoinOperator timeJoinOperator =
+ new TimeJoinOperator(
+ fragmentInstanceContext.getOperatorContexts().get(3),
+ Arrays.asList(seriesScanOperator1, seriesScanOperator2, seriesScanOperator3),
+ OrderBy.TIMESTAMP_ASC,
+ Arrays.asList(TSDataType.INT32, TSDataType.INT32, TSDataType.INT32),
+ Arrays.asList(
+ new SingleColumnMerger(new InputLocation(0, 0)),
+ new SingleColumnMerger(new InputLocation(1, 0)),
+ new SingleColumnMerger(new InputLocation(2, 0))));
+ int count = 0;
+ while (timeJoinOperator.hasNext()) {
+ TsBlock tsBlock = timeJoinOperator.next();
+ assertEquals(3, tsBlock.getValueColumnCount());
+ assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
+ assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
+ assertTrue(tsBlock.getColumn(2) instanceof RunLengthEncodedColumn);
+ assertEquals(20, tsBlock.getPositionCount());
+ for (int i = 0; i < tsBlock.getPositionCount(); i++) {
+ long expectedTime = i + 20L * count;
+ assertEquals(expectedTime, tsBlock.getTimeByIndex(i));
+ assertTrue(tsBlock.getColumn(2).isNull(i));
+ if (expectedTime < 200) {
+ assertEquals(20000 + expectedTime, tsBlock.getColumn(0).getInt(i));
+ assertEquals(20000 + expectedTime, tsBlock.getColumn(1).getInt(i));
+ } else if (expectedTime < 260
+ || (expectedTime >= 300 && expectedTime < 380)
+ || expectedTime >= 400) {
+ assertEquals(10000 + expectedTime, tsBlock.getColumn(0).getInt(i));
+ assertEquals(10000 + expectedTime, tsBlock.getColumn(1).getInt(i));
+ } else {
+ assertEquals(expectedTime, tsBlock.getColumn(0).getInt(i));
+ assertEquals(expectedTime, tsBlock.getColumn(1).getInt(i));
+ }
+ }
+ count++;
+ }
+ assertEquals(25, count);
+ } catch (IllegalPathException e) {
+ e.printStackTrace();
+ fail();
+ }
+ }
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/BinaryColumnBuilder.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/BinaryColumnBuilder.java
index 484b47b8e9..123a252c63 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/BinaryColumnBuilder.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/BinaryColumnBuilder.java
@@ -83,23 +83,13 @@ public class BinaryColumnBuilder implements ColumnBuilder {
}
@Override
- public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
- return writeBinary(value.getBinary());
+ public ColumnBuilder write(Column column, int index) {
+ return writeBinary(column.getBinary(index));
}
@Override
- public int appendColumn(
- TimeColumn timeColumn, Column valueColumn, int offset, TimeColumnBuilder timeBuilder) {
- int count = timeBuilder.getPositionCount();
- int index = offset;
- for (int i = 0; i < count; i++) {
- if (timeColumn.getPositionCount() > index && timeColumn.getLong(index) == timeBuilder.getTime(i) && !valueColumn.isNull(index)) {
- writeBinary(valueColumn.getBinary(index++));
- } else {
- appendNull();
- }
- }
- return index;
+ public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
+ return writeBinary(value.getBinary());
}
@Override
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/BooleanColumnBuilder.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/BooleanColumnBuilder.java
index 413747f72f..7632a5030b 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/BooleanColumnBuilder.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/BooleanColumnBuilder.java
@@ -85,23 +85,13 @@ public class BooleanColumnBuilder implements ColumnBuilder {
}
@Override
- public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
- return writeBoolean(value.getBoolean());
+ public ColumnBuilder write(Column column, int index) {
+ return writeBoolean(column.getBoolean(index));
}
@Override
- public int appendColumn(
- TimeColumn timeColumn, Column valueColumn, int offset, TimeColumnBuilder timeBuilder) {
- int count = timeBuilder.getPositionCount();
- int index = offset;
- for (int i = 0; i < count; i++) {
- if (timeColumn.getPositionCount() > index && timeColumn.getLong(index) == timeBuilder.getTime(i) && !valueColumn.isNull(index)) {
- writeBoolean(valueColumn.getBoolean(index++));
- } else {
- appendNull();
- }
- }
- return index;
+ public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
+ return writeBoolean(value.getBoolean());
}
@Override
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/ColumnBuilder.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/ColumnBuilder.java
index b431db1786..95e71edc06 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/ColumnBuilder.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/ColumnBuilder.java
@@ -64,12 +64,25 @@ public interface ColumnBuilder {
throw new UnsupportedOperationException(getClass().getName());
}
- int appendColumn(
- TimeColumn timeColumn, Column valueColumn, int offset, TimeColumnBuilder timeBuilder);
+ /**
+ * Write value at index of passing column
+ *
+ * @param column source column whose type should be same as ColumnBuilder
+ * @param index index of source column to read from
+ */
+ ColumnBuilder write(Column column, int index);
/** Appends a null value to the block. */
ColumnBuilder appendNull();
+ /** Appends nullCount null value to the block. */
+ default ColumnBuilder appendNull(int nullCount) {
+ for (int i = 0; i < nullCount; i++) {
+ appendNull();
+ }
+ return this;
+ }
+
/** Builds the block. This method can be called multiple times. */
Column build();
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/DoubleColumnBuilder.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/DoubleColumnBuilder.java
index 9ad7872283..0e5c98629a 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/DoubleColumnBuilder.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/DoubleColumnBuilder.java
@@ -85,23 +85,13 @@ public class DoubleColumnBuilder implements ColumnBuilder {
}
@Override
- public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
- return writeDouble(value.getDouble());
+ public ColumnBuilder write(Column column, int index) {
+ return writeDouble(column.getDouble(index));
}
@Override
- public int appendColumn(
- TimeColumn timeColumn, Column valueColumn, int offset, TimeColumnBuilder timeBuilder) {
- int count = timeBuilder.getPositionCount();
- int index = offset;
- for (int i = 0; i < count; i++) {
- if (timeColumn.getPositionCount() > index && timeColumn.getLong(index) == timeBuilder.getTime(i) && !valueColumn.isNull(index)) {
- writeDouble(valueColumn.getDouble(index++));
- } else {
- appendNull();
- }
- }
- return index;
+ public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
+ return writeDouble(value.getDouble());
}
@Override
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/FloatColumnBuilder.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/FloatColumnBuilder.java
index d609aaccd9..0d44550cae 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/FloatColumnBuilder.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/FloatColumnBuilder.java
@@ -90,18 +90,8 @@ public class FloatColumnBuilder implements ColumnBuilder {
}
@Override
- public int appendColumn(
- TimeColumn timeColumn, Column valueColumn, int offset, TimeColumnBuilder timeBuilder) {
- int count = timeBuilder.getPositionCount();
- int index = offset;
- for (int i = 0; i < count; i++) {
- if (timeColumn.getPositionCount() > index && timeColumn.getLong(index) == timeBuilder.getTime(i) && !valueColumn.isNull(index)) {
- writeFloat(valueColumn.getFloat(index++));
- } else {
- appendNull();
- }
- }
- return index;
+ public ColumnBuilder write(Column column, int index) {
+ return writeFloat(column.getFloat(index));
}
@Override
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/IntColumnBuilder.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/IntColumnBuilder.java
index 192c55f1e3..7593dc1dcd 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/IntColumnBuilder.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/IntColumnBuilder.java
@@ -85,23 +85,13 @@ public class IntColumnBuilder implements ColumnBuilder {
}
@Override
- public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
- return writeInt(value.getInt());
+ public ColumnBuilder write(Column column, int index) {
+ return writeInt(column.getInt(index));
}
@Override
- public int appendColumn(
- TimeColumn timeColumn, Column valueColumn, int offset, TimeColumnBuilder timeBuilder) {
- int count = timeBuilder.getPositionCount();
- int index = offset;
- for (int i = 0; i < count; i++) {
- if (timeColumn.getPositionCount() > index && timeColumn.getLong(index) == timeBuilder.getTime(i) && !valueColumn.isNull(index)) {
- writeInt(valueColumn.getInt(index++));
- } else {
- appendNull();
- }
- }
- return index;
+ public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
+ return writeInt(value.getInt());
}
@Override
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/LongColumnBuilder.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/LongColumnBuilder.java
index d29613e43d..0145049c3f 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/LongColumnBuilder.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/LongColumnBuilder.java
@@ -85,23 +85,13 @@ public class LongColumnBuilder implements ColumnBuilder {
}
@Override
- public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
- return writeLong(value.getLong());
+ public ColumnBuilder write(Column column, int index) {
+ return writeLong(column.getLong(index));
}
@Override
- public int appendColumn(
- TimeColumn timeColumn, Column valueColumn, int offset, TimeColumnBuilder timeBuilder) {
- int count = timeBuilder.getPositionCount();
- int index = offset;
- for (int i = 0; i < count; i++) {
- if (timeColumn.getPositionCount() > index && timeColumn.getLong(index) == timeBuilder.getTime(i) && !valueColumn.isNull(index)) {
- writeLong(valueColumn.getLong(index++));
- } else {
- appendNull();
- }
- }
- return index;
+ public ColumnBuilder writeTsPrimitiveType(TsPrimitiveType value) {
+ return writeLong(value.getLong());
}
@Override
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/TimeColumnBuilder.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/TimeColumnBuilder.java
index 99488d1213..59157916be 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/TimeColumnBuilder.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/block/column/TimeColumnBuilder.java
@@ -66,12 +66,6 @@ public class TimeColumnBuilder implements ColumnBuilder {
return this;
}
- @Override
- public int appendColumn(
- TimeColumn timeColumn, Column valueColumn, int offset, TimeColumnBuilder timeBuilder) {
- throw new UnsupportedOperationException(getClass().getName());
- }
-
/** Write an Object to the current entry, which should be the Long type; */
@Override
public ColumnBuilder writeObject(Object value) {
@@ -82,11 +76,21 @@ public class TimeColumnBuilder implements ColumnBuilder {
throw new UnSupportedDataTypeException("LongColumn only support Long data type");
}
+ @Override
+ public ColumnBuilder write(Column column, int index) {
+ return writeLong(column.getLong(index));
+ }
+
@Override
public ColumnBuilder appendNull() {
throw new UnsupportedOperationException(getClass().getName());
}
+ @Override
+ public ColumnBuilder appendNull(int nullCount) {
+ throw new UnsupportedOperationException(getClass().getName());
+ }
+
@Override
public Column build() {
return new TimeColumn(0, positionCount, values);