| /* |
| * 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.queryengine.execution.operator.process.join; |
| |
| import org.apache.iotdb.db.queryengine.execution.operator.Operator; |
| import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext; |
| import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.AscTimeComparator; |
| import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.DescTimeComparator; |
| import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation; |
| import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; |
| 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.utils.Binary; |
| |
| import com.google.common.util.concurrent.ListenableFuture; |
| import io.airlift.units.Duration; |
| import org.junit.Test; |
| import org.mockito.Mockito; |
| |
| import java.nio.charset.StandardCharsets; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.Map; |
| import java.util.concurrent.TimeUnit; |
| |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| public class InnerTimeJoinOperatorTest { |
| |
| @Test |
| public void testInnerJoin1() { |
| // left table |
| // Time, s1 |
| // 4 4 |
| // 6 6 |
| // 9 9 |
| // ----------- TsBlock-1 |
| // 13 13 |
| // 17 17 |
| // ----------- TsBlock-2 |
| // 22 22 |
| // 25 25 |
| // ----------- TsBlock-3 |
| |
| // right table |
| // Time, s2 |
| // 1 10 |
| // 2 20 |
| // 3 30 |
| // ----------- TsBlock-1 |
| // 4 40 |
| // 5 50 |
| // 10 100 |
| // ----------- TsBlock-2 |
| // 13 130 |
| // 16 160 |
| // ----------- TsBlock-3 |
| // 26 260 |
| // 27 270 |
| // ----------- TsBlock-4 |
| |
| // result table |
| // Time, s1, s2 |
| // 4, 4, 40 |
| // 13 13, 130 |
| |
| OperatorContext operatorContext = Mockito.mock(OperatorContext.class); |
| Mockito.when(operatorContext.getMaxRunTime()).thenReturn(new Duration(1, TimeUnit.SECONDS)); |
| |
| Operator leftChild = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] { |
| {4L, 6L, 9L}, |
| {13L, 17L}, |
| {22L, 25L} |
| }; |
| |
| private final int[][] valueArray = |
| new int[][] { |
| {4, 6, 9}, |
| {13, 17}, |
| {22, 25} |
| }; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| { |
| {false, false, false}, |
| {false, false}, |
| {false, false} |
| } |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Collections.singletonList(TSDataType.INT32)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeInt(valueArray[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 3; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 3; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Operator rightChild = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] { |
| {1L, 2L, 3L}, |
| {4L, 5L, 10L}, |
| {13L, 16L}, |
| {26L, 27L} |
| }; |
| |
| private final long[][] valueArray = |
| new long[][] { |
| {10L, 20L, 30L}, |
| {40L, 50L, 100L}, |
| {130L, 160L}, |
| {260L, 270L} |
| }; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| { |
| {false, false, false}, |
| {false, false, false}, |
| {false, false}, |
| {false, false} |
| } |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Collections.singletonList(TSDataType.INT64)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeLong(valueArray[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 4; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 4; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Map<InputLocation, Integer> outputColumnMap = new HashMap<>(); |
| outputColumnMap.put(new InputLocation(0, 0), 0); |
| outputColumnMap.put(new InputLocation(1, 0), 1); |
| |
| InnerTimeJoinOperator innerTimeJoinOperator = |
| new InnerTimeJoinOperator( |
| operatorContext, |
| Arrays.asList(leftChild, rightChild), |
| Arrays.asList(TSDataType.INT32, TSDataType.INT64), |
| new AscTimeComparator(), |
| outputColumnMap); |
| |
| assertEquals( |
| TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes() + 64 * 1024 * 2, |
| innerTimeJoinOperator.calculateMaxPeekMemory()); |
| assertEquals( |
| TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes(), |
| innerTimeJoinOperator.calculateMaxReturnSize()); |
| assertEquals(64 * 1024, innerTimeJoinOperator.calculateRetainedSizeAfterCallingNext()); |
| |
| long[] timeArray = new long[] {4L, 13L}; |
| int[] column1Array = new int[] {4, 13}; |
| boolean[] column1IsNull = new boolean[] {false, false}; |
| long[] column2Array = new long[] {40L, 130L}; |
| boolean[] column2IsNull = new boolean[] {false, false}; |
| |
| try { |
| int count = 0; |
| ListenableFuture<?> listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| while (!innerTimeJoinOperator.isFinished() && innerTimeJoinOperator.hasNext()) { |
| TsBlock tsBlock = innerTimeJoinOperator.next(); |
| if (tsBlock != null && !tsBlock.isEmpty()) { |
| for (int i = 0, size = tsBlock.getPositionCount(); i < size; i++, count++) { |
| assertEquals(timeArray[count], tsBlock.getTimeByIndex(i)); |
| assertEquals(column1IsNull[count], tsBlock.getColumn(0).isNull(i)); |
| if (!column1IsNull[count]) { |
| assertEquals(column1Array[count], tsBlock.getColumn(0).getInt(i)); |
| } |
| assertEquals(column2IsNull[count], tsBlock.getColumn(1).isNull(i)); |
| if (!column2IsNull[count]) { |
| assertEquals(column2Array[count], tsBlock.getColumn(1).getLong(i)); |
| } |
| } |
| } |
| listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| } |
| assertEquals(timeArray.length, count); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail(e.getMessage()); |
| } |
| } |
| |
| @Test |
| public void testInnerJoin2() { |
| // left table |
| // Time, s1, s2 |
| // 25 null 26 |
| // 22 22 null |
| // ---------------------- TsBlock-1 |
| // null |
| // ---------------------- TsBlock-2 |
| // 19 19 20 |
| // 18 18 null |
| // 15 null 16 |
| // ---------------------- TsBlock-3 |
| // empty |
| // ---------------------- TsBlock-4 |
| // 9 null null |
| // 7 7 null |
| // 6 null 7 |
| // 3 3 4 |
| // ---------------------- TsBlock-5 |
| // empty |
| // ---------------------- TsBlock-6 |
| |
| // right table |
| // Time, s3, s4 |
| // 21 210.0 false |
| // 20 200.0 null |
| // ---------------------- TsBlock-1 |
| // empty |
| // ---------------------- TsBlock-2 |
| // 19 190.0 true |
| // 18 180.0 null |
| // 15 null false |
| // 14 null null |
| // 8 80.0 true |
| // 7 null false |
| // ---------------------- TsBlock-3 |
| // null |
| // ---------------------- TsBlock-4 |
| // 5 50.0 true |
| // ---------------------- TsBlock-5 |
| // 4 40.0 null |
| // ---------------------- TsBlock-6 |
| // 3 30.0 false |
| // ---------------------- TsBlock-7 |
| // 2 20.0 true |
| // 1 10.0 false |
| // ---------------------- TsBlock-8 |
| // empty |
| // ---------------------- TsBlock-9 |
| |
| // result table |
| // Time, s1, s2, s3, s4 |
| // 19 19 20 190.0 true |
| // 18 18 null 180.0 null |
| // 15 null 16 null false |
| // 7 7 null null false |
| // 3 3 4 30.0 false |
| |
| OperatorContext operatorContext = Mockito.mock(OperatorContext.class); |
| Mockito.when(operatorContext.getMaxRunTime()).thenReturn(new Duration(1000, TimeUnit.SECONDS)); |
| |
| Operator leftChild = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] {{25L, 22L}, null, {19L, 18L, 15L}, {}, {9L, 7L, 6L, 3L}, {}}; |
| |
| private final int[][] value1Array = |
| new int[][] {{0, 22}, null, {19, 18, 0}, {}, {0, 7, 0, 3}, {}}; |
| |
| private final long[][] value2Array = |
| new long[][] {{26L, 0L}, null, {20L, 0L, 16L}, {}, {0L, 0L, 7L, 4L}, {}}; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| {{true, false}, null, {false, false, true}, {}, {true, false, true, false}, {}}, |
| {{false, true}, null, {false, true, false}, {}, {true, true, false, false}, {}} |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Arrays.asList(TSDataType.INT32, TSDataType.INT64)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeInt(value1Array[index][i]); |
| } |
| if (valueIsNull[1][index][i]) { |
| builder.getColumnBuilder(1).appendNull(); |
| } else { |
| builder.getColumnBuilder(1).writeLong(value2Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 6; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 6; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Operator rightChild = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] { |
| {21L, 20L}, {}, {19L, 18L, 15L, 14L, 8L, 7L}, null, {5L}, {4L}, {3L}, {2L, 1L}, {} |
| }; |
| |
| private final float[][] value1Array = |
| new float[][] { |
| {210.0f, 200.0f}, |
| {}, |
| {190.0f, 180.0f, 0.0f, 0.0f, 80.0f, 0.0f}, |
| null, |
| {50.0f}, |
| {40.0f}, |
| {30.0f}, |
| {20.0f, 10.0f}, |
| {} |
| }; |
| |
| private final boolean[][] value2Array = |
| new boolean[][] { |
| {false, false}, |
| {}, |
| {true, false, false, false, true, false}, |
| null, |
| {true}, |
| {false}, |
| {false}, |
| {true, false}, |
| {} |
| }; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| { |
| {false, false}, |
| {}, |
| {false, false, true, true, false, true}, |
| null, |
| {false}, |
| {false}, |
| {false}, |
| {false, false}, |
| {} |
| }, |
| { |
| {false, true}, |
| {}, |
| {false, true, false, true, false, false}, |
| null, |
| {false}, |
| {true}, |
| {false}, |
| {false, false}, |
| {} |
| } |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeFloat(value1Array[index][i]); |
| } |
| if (valueIsNull[1][index][i]) { |
| builder.getColumnBuilder(1).appendNull(); |
| } else { |
| builder.getColumnBuilder(1).writeBoolean(value2Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 9; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 9; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Map<InputLocation, Integer> outputColumnMap = new HashMap<>(); |
| outputColumnMap.put(new InputLocation(0, 0), 0); |
| outputColumnMap.put(new InputLocation(0, 1), 1); |
| outputColumnMap.put(new InputLocation(1, 0), 2); |
| outputColumnMap.put(new InputLocation(1, 1), 3); |
| |
| InnerTimeJoinOperator innerTimeJoinOperator = |
| new InnerTimeJoinOperator( |
| operatorContext, |
| Arrays.asList(leftChild, rightChild), |
| Arrays.asList(TSDataType.INT32, TSDataType.INT64, TSDataType.FLOAT, TSDataType.BOOLEAN), |
| new DescTimeComparator(), |
| outputColumnMap); |
| |
| long[] timeArray = new long[] {19L, 18L, 15L, 7L, 3L}; |
| int[] column1Array = new int[] {19, 18, 0, 7, 3}; |
| boolean[] column1IsNull = new boolean[] {false, false, true, false, false}; |
| long[] column2Array = new long[] {20L, 0L, 16L, 0L, 4L}; |
| boolean[] column2IsNull = new boolean[] {false, true, false, true, false}; |
| float[] column3Array = new float[] {190.0f, 180.0f, 0.0f, 0.0f, 30.0f}; |
| boolean[] column3IsNull = new boolean[] {false, false, true, true, false}; |
| boolean[] column4Array = new boolean[] {true, false, false, false, false}; |
| boolean[] column4IsNull = new boolean[] {false, true, false, false, false}; |
| |
| try { |
| int count = 0; |
| ListenableFuture<?> listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| while (!innerTimeJoinOperator.isFinished() && innerTimeJoinOperator.hasNext()) { |
| TsBlock tsBlock = innerTimeJoinOperator.next(); |
| if (tsBlock != null && !tsBlock.isEmpty()) { |
| for (int i = 0, size = tsBlock.getPositionCount(); i < size; i++, count++) { |
| assertEquals(timeArray[count], tsBlock.getTimeByIndex(i)); |
| assertEquals(column1IsNull[count], tsBlock.getColumn(0).isNull(i)); |
| if (!column1IsNull[count]) { |
| assertEquals(column1Array[count], tsBlock.getColumn(0).getInt(i)); |
| } |
| assertEquals(column2IsNull[count], tsBlock.getColumn(1).isNull(i)); |
| if (!column2IsNull[count]) { |
| assertEquals(column2Array[count], tsBlock.getColumn(1).getLong(i)); |
| } |
| assertEquals(column3IsNull[count], tsBlock.getColumn(2).isNull(i)); |
| if (!column3IsNull[count]) { |
| assertEquals(column3Array[count], tsBlock.getColumn(2).getFloat(i), 0.000001); |
| } |
| assertEquals(column4IsNull[count], tsBlock.getColumn(3).isNull(i)); |
| if (!column4IsNull[count]) { |
| assertEquals(column4Array[count], tsBlock.getColumn(3).getBoolean(i)); |
| } |
| } |
| } |
| listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| } |
| assertEquals(timeArray.length, count); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail(e.getMessage()); |
| } |
| } |
| |
| @Test |
| public void testInnerJoin3() { |
| // child-1 |
| // Time, s1, s2 |
| // 100 100 200 |
| // 90 90 180 |
| // 80 80 160 |
| // 70 70 140 |
| // 60 60 120 |
| // 50 50 100 |
| // 40 40 80 |
| // 30 30 60 |
| // 20 20 40 |
| // 10 10 20 |
| // 0 0 0 |
| // ---------------------- TsBlock-1 |
| |
| // child-2 |
| // Time, s3, s4 |
| // 1000 3000.0 false |
| // 500 500.0 true |
| // 100 300.0 null |
| // ------------------------- TsBlock-1 |
| // 99 99.0 true |
| // 95 95.0 null |
| // 90 null false |
| // ------------------------- TsBlock-2 |
| // 50 150.0 true |
| // 48 48.0 true |
| // 20 60.0 null |
| // 10 null false |
| // ------------------------- TsBlock-3 |
| |
| // result table |
| // Time, s1, s2, s3, s4 |
| // 100 100 200 300.0 null |
| // 90 90 180 null false |
| // 50 50 100 150.0 true |
| // 20 20 40 60.0 null |
| // 10 10 20 null false |
| |
| OperatorContext operatorContext = Mockito.mock(OperatorContext.class); |
| Mockito.when(operatorContext.getMaxRunTime()).thenReturn(new Duration(1000, TimeUnit.SECONDS)); |
| |
| Operator child1 = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] {{100L, 90L, 80L, 70L, 60L, 50L, 40L, 30L, 20L, 10L, 0L}}; |
| |
| private final int[][] value1Array = |
| new int[][] {{100, 90, 80, 70, 60, 50, 40, 30, 20, 10, 0}}; |
| |
| private final long[][] value2Array = |
| new long[][] {{200L, 180L, 160, 140, 120, 100L, 80L, 60L, 40L, 20L, 0L}}; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| {{false, false, false, false, false, false, false, false, false, false, false}}, |
| {{false, false, false, false, false, false, false, false, false, false, false}} |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Arrays.asList(TSDataType.INT32, TSDataType.INT64)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeInt(value1Array[index][i]); |
| } |
| if (valueIsNull[1][index][i]) { |
| builder.getColumnBuilder(1).appendNull(); |
| } else { |
| builder.getColumnBuilder(1).writeLong(value2Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 1; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 1; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Operator child2 = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] {{1000L, 500L, 100L}, {99L, 95L, 90L}, {50L, 48L, 20L, 10L}}; |
| |
| private final float[][] value1Array = |
| new float[][] { |
| {3000.0f, 500.0f, 300.0f}, |
| {99.0f, 95.0f, 0.0f}, |
| {150.0f, 48.0f, 60.0f, 0.0f} |
| }; |
| |
| private final boolean[][] value2Array = |
| new boolean[][] { |
| {false, true, false}, |
| {true, false, false}, |
| {true, true, false, false} |
| }; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| { |
| {false, false, false}, |
| {false, false, true}, |
| {false, false, false, true} |
| }, |
| { |
| {false, false, true}, |
| {false, true, false}, |
| {false, false, true, false} |
| } |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeFloat(value1Array[index][i]); |
| } |
| if (valueIsNull[1][index][i]) { |
| builder.getColumnBuilder(1).appendNull(); |
| } else { |
| builder.getColumnBuilder(1).writeBoolean(value2Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 3; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 3; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Map<InputLocation, Integer> outputColumnMap = new HashMap<>(); |
| outputColumnMap.put(new InputLocation(0, 0), 0); |
| outputColumnMap.put(new InputLocation(0, 1), 1); |
| outputColumnMap.put(new InputLocation(1, 0), 2); |
| outputColumnMap.put(new InputLocation(1, 1), 3); |
| |
| InnerTimeJoinOperator innerTimeJoinOperator = |
| new InnerTimeJoinOperator( |
| operatorContext, |
| Arrays.asList(child1, child2), |
| Arrays.asList(TSDataType.INT32, TSDataType.INT64, TSDataType.FLOAT, TSDataType.BOOLEAN), |
| new DescTimeComparator(), |
| outputColumnMap); |
| |
| long[] timeArray = new long[] {100L, 90L, 50L, 20L, 10L}; |
| int[] column1Array = new int[] {100, 90, 50, 20, 10}; |
| boolean[] column1IsNull = new boolean[] {false, false, false, false, false}; |
| long[] column2Array = new long[] {200L, 180L, 100L, 40L, 20L}; |
| boolean[] column2IsNull = new boolean[] {false, false, false, false, false}; |
| float[] column3Array = new float[] {300.0f, 0.0f, 150.0f, 60.0f, 0.0f}; |
| boolean[] column3IsNull = new boolean[] {false, true, false, false, true}; |
| boolean[] column4Array = new boolean[] {false, false, true, false, false}; |
| boolean[] column4IsNull = new boolean[] {true, false, false, true, false}; |
| |
| try { |
| int count = 0; |
| ListenableFuture<?> listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| while (!innerTimeJoinOperator.isFinished() && innerTimeJoinOperator.hasNext()) { |
| TsBlock tsBlock = innerTimeJoinOperator.next(); |
| if (tsBlock != null && !tsBlock.isEmpty()) { |
| for (int i = 0, size = tsBlock.getPositionCount(); i < size; i++, count++) { |
| assertEquals(timeArray[count], tsBlock.getTimeByIndex(i)); |
| assertEquals(column1IsNull[count], tsBlock.getColumn(0).isNull(i)); |
| if (!column1IsNull[count]) { |
| assertEquals(column1Array[count], tsBlock.getColumn(0).getInt(i)); |
| } |
| assertEquals(column2IsNull[count], tsBlock.getColumn(1).isNull(i)); |
| if (!column2IsNull[count]) { |
| assertEquals(column2Array[count], tsBlock.getColumn(1).getLong(i)); |
| } |
| assertEquals(column3IsNull[count], tsBlock.getColumn(2).isNull(i)); |
| if (!column3IsNull[count]) { |
| assertEquals(column3Array[count], tsBlock.getColumn(2).getFloat(i), 0.000001); |
| } |
| assertEquals(column4IsNull[count], tsBlock.getColumn(3).isNull(i)); |
| if (!column4IsNull[count]) { |
| assertEquals(column4Array[count], tsBlock.getColumn(3).getBoolean(i)); |
| } |
| } |
| } |
| listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| } |
| assertEquals(timeArray.length, count); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail(e.getMessage()); |
| } |
| } |
| |
| @Test |
| public void testInnerJoin4() { |
| // child-1 |
| // Time, s1, s2 |
| // 100 100 200 |
| // 90 90 180 |
| // 80 80 160 |
| // 70 70 140 |
| // 60 60 120 |
| // 50 50 100 |
| // 40 40 80 |
| // 30 30 60 |
| // 20 20 40 |
| // 10 10 20 |
| // 0 0 0 |
| // ---------------------- TsBlock-1 |
| |
| // child-2 |
| // Time, s3, s4 |
| // 1000 3000.0 false |
| // 500 500.0 true |
| // 100 300.0 null |
| // ------------------------- TsBlock-1 |
| // 99 99.0 true |
| // 95 95.0 null |
| // 90 null false |
| // ------------------------- TsBlock-2 |
| // 50 150.0 true |
| // 48 48.0 true |
| // 20 60.0 null |
| // 10 null false |
| // ------------------------- TsBlock-3 |
| |
| // child-3 |
| // Time, s5, |
| // 1000 "iotdb" |
| // 500 "ty" |
| // 101 "zm" |
| // --------------------- TsBlock-1 |
| // 99 "ty" |
| // 95 "love" |
| // 80 "zm" |
| // 60 "2018-05-06" |
| // --------------------- TsBlock-2 |
| // 40 "1997-09-09" |
| // 22 "1995-04-21" |
| // 11 "2022-04-21" |
| // 0 "2023-12-30" |
| // --------------------- TsBlock-3 |
| |
| // result table |
| // Time, s1, s2, s3, s4, s5 |
| // empty |
| |
| OperatorContext operatorContext = Mockito.mock(OperatorContext.class); |
| Mockito.when(operatorContext.getMaxRunTime()).thenReturn(new Duration(1000, TimeUnit.SECONDS)); |
| |
| Operator child1 = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] {{100L, 90L, 80L, 70L, 60L, 50L, 40L, 30L, 20L, 10L, 0L}}; |
| |
| private final int[][] value1Array = |
| new int[][] {{100, 90, 80, 70, 60, 50, 40, 30, 20, 10, 0}}; |
| |
| private final long[][] value2Array = |
| new long[][] {{200L, 180L, 160, 140, 120, 100L, 80L, 60L, 40L, 20L, 0L}}; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| {{false, false, false, false, false, false, false, false, false, false, false}}, |
| {{false, false, false, false, false, false, false, false, false, false, false}} |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Arrays.asList(TSDataType.INT32, TSDataType.INT64)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeInt(value1Array[index][i]); |
| } |
| if (valueIsNull[1][index][i]) { |
| builder.getColumnBuilder(1).appendNull(); |
| } else { |
| builder.getColumnBuilder(1).writeLong(value2Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 1; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 1; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Operator child2 = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] {{1000L, 500L, 100L}, {99L, 95L, 90L}, {50L, 48L, 20L, 10L}}; |
| |
| private final float[][] value1Array = |
| new float[][] { |
| {3000.0f, 500.0f, 300.0f}, |
| {99.0f, 95.0f, 0.0f}, |
| {150.0f, 48.0f, 60.0f, 0.0f} |
| }; |
| |
| private final boolean[][] value2Array = |
| new boolean[][] { |
| {false, true, false}, |
| {true, false, false}, |
| {true, true, false, false} |
| }; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| { |
| {false, false, false}, |
| {false, false, true}, |
| {false, false, false, true} |
| }, |
| { |
| {false, false, true}, |
| {false, true, false}, |
| {false, false, true, false} |
| } |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeFloat(value1Array[index][i]); |
| } |
| if (valueIsNull[1][index][i]) { |
| builder.getColumnBuilder(1).appendNull(); |
| } else { |
| builder.getColumnBuilder(1).writeBoolean(value2Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 3; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 3; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Operator child3 = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] {{1000L, 500L, 101L}, {99L, 95L, 80L, 60L}, {40L, 22L, 11L, 0L}}; |
| |
| private final Binary[][] value1Array = |
| new Binary[][] { |
| { |
| new Binary("iotdb".getBytes(StandardCharsets.UTF_8)), |
| new Binary("ty".getBytes(StandardCharsets.UTF_8)), |
| new Binary("zm".getBytes(StandardCharsets.UTF_8)) |
| }, |
| { |
| new Binary("ty".getBytes(StandardCharsets.UTF_8)), |
| new Binary("love".getBytes(StandardCharsets.UTF_8)), |
| new Binary("zm".getBytes(StandardCharsets.UTF_8)), |
| new Binary("2018-05-06".getBytes(StandardCharsets.UTF_8)) |
| }, |
| { |
| new Binary("1997-09-09".getBytes(StandardCharsets.UTF_8)), |
| new Binary("1995-04-21".getBytes(StandardCharsets.UTF_8)), |
| new Binary("2022-04-21".getBytes(StandardCharsets.UTF_8)), |
| new Binary("2023-12-30".getBytes(StandardCharsets.UTF_8)) |
| } |
| }; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| { |
| {false, false, false}, |
| {false, false, false, false}, |
| {false, false, false, false} |
| } |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder(timeArray[index].length, Arrays.asList(TSDataType.TEXT)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeBinary(value1Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 3; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 3; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Map<InputLocation, Integer> outputColumnMap = new HashMap<>(); |
| outputColumnMap.put(new InputLocation(0, 0), 0); |
| outputColumnMap.put(new InputLocation(0, 1), 1); |
| outputColumnMap.put(new InputLocation(1, 0), 2); |
| outputColumnMap.put(new InputLocation(1, 1), 3); |
| outputColumnMap.put(new InputLocation(2, 0), 4); |
| |
| InnerTimeJoinOperator innerTimeJoinOperator = |
| new InnerTimeJoinOperator( |
| operatorContext, |
| Arrays.asList(child1, child2, child3), |
| Arrays.asList( |
| TSDataType.INT32, |
| TSDataType.INT64, |
| TSDataType.FLOAT, |
| TSDataType.BOOLEAN, |
| TSDataType.TEXT), |
| new DescTimeComparator(), |
| outputColumnMap); |
| |
| try { |
| int count = 0; |
| ListenableFuture<?> listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| while (!innerTimeJoinOperator.isFinished() && innerTimeJoinOperator.hasNext()) { |
| TsBlock tsBlock = innerTimeJoinOperator.next(); |
| if (tsBlock != null && !tsBlock.isEmpty()) { |
| count += tsBlock.getPositionCount(); |
| } |
| listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| } |
| assertEquals(0, count); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail(e.getMessage()); |
| } |
| } |
| |
| @Test |
| public void testInnerJoin5() { |
| // child-1 |
| // Time, s1, s2 |
| // 100 100 200 |
| // 90 90 180 |
| // 80 80 160 |
| // 70 70 140 |
| // 60 60 120 |
| // 50 50 100 |
| // 40 40 80 |
| // 30 30 60 |
| // 20 20 40 |
| // 10 10 20 |
| // 0 0 0 |
| // ---------------------- TsBlock-1 |
| |
| // child-2 |
| // Time, s3, s4 |
| // 1000 3000.0 false |
| // 500 500.0 true |
| // 100 300.0 null |
| // ------------------------- TsBlock-1 |
| // 99 99.0 true |
| // 95 95.0 null |
| // 90 null false |
| // ------------------------- TsBlock-2 |
| // 50 150.0 true |
| // 48 48.0 true |
| // 20 60.0 null |
| // 10 null false |
| // ------------------------- TsBlock-3 |
| |
| // child-3 |
| // Time, s5, |
| // 1000 "iotdb" |
| // 500 "ty" |
| // 101 "zm" |
| // --------------------- TsBlock-1 |
| // 99 "ty" |
| // 90 "love" |
| // 80 "zm" |
| // 60 "2018-05-06" |
| // --------------------- TsBlock-2 |
| // 40 "1997-09-09" |
| // 22 "1995-04-21" |
| // 11 "2022-04-21" |
| // 0 "2023-12-30" |
| // --------------------- TsBlock-3 |
| |
| // result table |
| // Time, s1, s2, s3, s4, s5 |
| // 90 90 180 null false "love" |
| |
| OperatorContext operatorContext = Mockito.mock(OperatorContext.class); |
| Mockito.when(operatorContext.getMaxRunTime()).thenReturn(new Duration(1000, TimeUnit.SECONDS)); |
| |
| Operator child1 = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] {{100L, 90L, 80L, 70L, 60L, 50L, 40L, 30L, 20L, 10L, 0L}}; |
| |
| private final int[][] value1Array = |
| new int[][] {{100, 90, 80, 70, 60, 50, 40, 30, 20, 10, 0}}; |
| |
| private final long[][] value2Array = |
| new long[][] {{200L, 180L, 160, 140, 120, 100L, 80L, 60L, 40L, 20L, 0L}}; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| {{false, false, false, false, false, false, false, false, false, false, false}}, |
| {{false, false, false, false, false, false, false, false, false, false, false}} |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Arrays.asList(TSDataType.INT32, TSDataType.INT64)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeInt(value1Array[index][i]); |
| } |
| if (valueIsNull[1][index][i]) { |
| builder.getColumnBuilder(1).appendNull(); |
| } else { |
| builder.getColumnBuilder(1).writeLong(value2Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 1; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 1; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Operator child2 = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] {{1000L, 500L, 100L}, {99L, 95L, 90L}, {50L, 48L, 20L, 10L}}; |
| |
| private final float[][] value1Array = |
| new float[][] { |
| {3000.0f, 500.0f, 300.0f}, |
| {99.0f, 95.0f, 0.0f}, |
| {150.0f, 48.0f, 60.0f, 0.0f} |
| }; |
| |
| private final boolean[][] value2Array = |
| new boolean[][] { |
| {false, true, false}, |
| {true, false, false}, |
| {true, true, false, false} |
| }; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| { |
| {false, false, false}, |
| {false, false, true}, |
| {false, false, false, true} |
| }, |
| { |
| {false, false, true}, |
| {false, true, false}, |
| {false, false, true, false} |
| } |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder( |
| timeArray[index].length, Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeFloat(value1Array[index][i]); |
| } |
| if (valueIsNull[1][index][i]) { |
| builder.getColumnBuilder(1).appendNull(); |
| } else { |
| builder.getColumnBuilder(1).writeBoolean(value2Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 3; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 3; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Operator child3 = |
| new Operator() { |
| private final long[][] timeArray = |
| new long[][] {{1000L, 500L, 101L}, {99L, 90L, 80L, 60L}, {40L, 22L, 11L, 0L}}; |
| |
| private final Binary[][] value1Array = |
| new Binary[][] { |
| { |
| new Binary("iotdb".getBytes(StandardCharsets.UTF_8)), |
| new Binary("ty".getBytes(StandardCharsets.UTF_8)), |
| new Binary("zm".getBytes(StandardCharsets.UTF_8)) |
| }, |
| { |
| new Binary("ty".getBytes(StandardCharsets.UTF_8)), |
| new Binary("love".getBytes(StandardCharsets.UTF_8)), |
| new Binary("zm".getBytes(StandardCharsets.UTF_8)), |
| new Binary("2018-05-06".getBytes(StandardCharsets.UTF_8)) |
| }, |
| { |
| new Binary("1997-09-09".getBytes(StandardCharsets.UTF_8)), |
| new Binary("1995-04-21".getBytes(StandardCharsets.UTF_8)), |
| new Binary("2022-04-21".getBytes(StandardCharsets.UTF_8)), |
| new Binary("2023-12-30".getBytes(StandardCharsets.UTF_8)) |
| } |
| }; |
| |
| private final boolean[][][] valueIsNull = |
| new boolean[][][] { |
| { |
| {false, false, false}, |
| {false, false, false, false}, |
| {false, false, false, false} |
| } |
| }; |
| |
| private int index = 0; |
| |
| @Override |
| public OperatorContext getOperatorContext() { |
| return operatorContext; |
| } |
| |
| @Override |
| public TsBlock next() { |
| if (timeArray[index] == null) { |
| index++; |
| return null; |
| } |
| TsBlockBuilder builder = |
| new TsBlockBuilder(timeArray[index].length, Arrays.asList(TSDataType.TEXT)); |
| for (int i = 0, size = timeArray[index].length; i < size; i++) { |
| builder.getTimeColumnBuilder().writeLong(timeArray[index][i]); |
| if (valueIsNull[0][index][i]) { |
| builder.getColumnBuilder(0).appendNull(); |
| } else { |
| builder.getColumnBuilder(0).writeBinary(value1Array[index][i]); |
| } |
| } |
| builder.declarePositions(timeArray[index].length); |
| index++; |
| return builder.build(); |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return index < 3; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public boolean isFinished() { |
| return index >= 3; |
| } |
| |
| @Override |
| public long calculateMaxPeekMemory() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateMaxReturnSize() { |
| return 64 * 1024; |
| } |
| |
| @Override |
| public long calculateRetainedSizeAfterCallingNext() { |
| return 0; |
| } |
| }; |
| |
| Map<InputLocation, Integer> outputColumnMap = new HashMap<>(); |
| outputColumnMap.put(new InputLocation(0, 0), 0); |
| outputColumnMap.put(new InputLocation(0, 1), 1); |
| outputColumnMap.put(new InputLocation(1, 0), 2); |
| outputColumnMap.put(new InputLocation(1, 1), 3); |
| outputColumnMap.put(new InputLocation(2, 0), 4); |
| |
| InnerTimeJoinOperator innerTimeJoinOperator = |
| new InnerTimeJoinOperator( |
| operatorContext, |
| Arrays.asList(child1, child2, child3), |
| Arrays.asList( |
| TSDataType.INT32, |
| TSDataType.INT64, |
| TSDataType.FLOAT, |
| TSDataType.BOOLEAN, |
| TSDataType.TEXT), |
| new DescTimeComparator(), |
| outputColumnMap); |
| |
| try { |
| int count = 0; |
| ListenableFuture<?> listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| while (!innerTimeJoinOperator.isFinished() && innerTimeJoinOperator.hasNext()) { |
| TsBlock tsBlock = innerTimeJoinOperator.next(); |
| if (tsBlock != null && !tsBlock.isEmpty()) { |
| count += tsBlock.getPositionCount(); |
| assertEquals(90, tsBlock.getTimeByIndex(0)); |
| assertFalse(tsBlock.getColumn(0).isNull(0)); |
| assertEquals(90, tsBlock.getColumn(0).getInt(0)); |
| assertFalse(tsBlock.getColumn(1).isNull(0)); |
| assertEquals(180L, tsBlock.getColumn(1).getLong(0)); |
| assertTrue(tsBlock.getColumn(2).isNull(0)); |
| assertFalse(tsBlock.getColumn(3).isNull(0)); |
| assertFalse(tsBlock.getColumn(3).getBoolean(0)); |
| assertFalse(tsBlock.getColumn(4).isNull(0)); |
| assertEquals( |
| "love", tsBlock.getColumn(4).getBinary(0).getStringValue(StandardCharsets.UTF_8)); |
| } |
| listenableFuture = innerTimeJoinOperator.isBlocked(); |
| listenableFuture.get(); |
| } |
| assertEquals(1, count); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail(e.getMessage()); |
| } |
| } |
| } |
