Vladsz83 commented on code in PR #13407:
URL: https://github.com/apache/ignite/pull/13407#discussion_r3673884812
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java:
##########
@@ -155,6 +155,30 @@ public void testNullsReordering() {
.check();
}
+ /**
+ * Tests that sort aggregate node correctly handles the case when input
data
+ * ends exactly when the requested number of rows is satisfied.
+ */
+ @Test
+ public void testRequestRowsAfterInputEnds() {
Review Comment:
It actually works witout fixes in the `ALL` and `RANDOM` tx. modes. Maybe we
need `assumeThat()`
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CollectNode.java:
##########
@@ -114,6 +114,11 @@ public static <Row> CollectNode<Row>
createCountCollector(ExecutionContext<Row>
if (waiting == 0)
source().request(waiting = IN_BUFFER_SIZE);
Review Comment:
Brw. Why di we reques more than even `rowsCnt` (in case of few `rowsCnt`)?
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java:
##########
@@ -155,6 +155,30 @@ public void testNullsReordering() {
.check();
}
+ /**
+ * Tests that sort aggregate node correctly handles the case when input
data
+ * ends exactly when the requested number of rows is satisfied.
+ */
+ @Test
+ public void testRequestRowsAfterInputEnds() {
+ /**
+ * With 512 input rows (equal to buffer size), the last row completes
both
+ * the input data and the requested count in the same cycle. This
triggers
+ * a synchronous request() call from within push() to fill the buffer,
and
+ * the node must properly handle the termination on the subsequent
request()
+ * call rather than on end().
+ */
+ int bufSize = 512;
Review Comment:
`AbstractNode#IN_BUFFER_SIZE` maybe?
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CollectIntegrationTest.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.integration;
+
+import org.junit.Test;
+
+/**
+ * Integration test for collect node.
+ */
+public class CollectIntegrationTest extends AbstractBasicIntegrationTest {
+ /**
+ * Tests that collect node correctly handles the case when downstream
requests
+ * limited number of rows, where collect must push one row and then
+ * properly terminate downstream.
+ */
+ @Test
+ public void testRequestLimitedRowsCountFromCollect() {
+ sql("CREATE TABLE t(a INT)");
+
+ sql("INSERT INTO t (a) VALUES (?)", 0);
+
+ String sql = "SELECT /*+ CNL_JOIN */ ARRAY(SELECT a FROM t) FROM t
LIMIT 1";
+
+ assertQuery(sql).resultSize(1).check();
+
+ /**
+ * The data source size of 513 (buffer size + 1) is used to ensure
that multiple batches are needed
+ * on right hand of CNLJ to process all input rows, in this case left
hand is not requested
+ * immediately after endLeft() call.
+ */
+ for (int i = 1; i < 513; i++)
Review Comment:
`AbstractNode#IN_BUFFER_SIZE` maybe? Or ar least a variable/constant?
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java:
##########
@@ -155,6 +155,30 @@ public void testNullsReordering() {
.check();
}
+ /**
+ * Tests that sort aggregate node correctly handles the case when input
data
+ * ends exactly when the requested number of rows is satisfied.
+ */
+ @Test
+ public void testRequestRowsAfterInputEnds() {
+ /**
+ * With 512 input rows (equal to buffer size), the last row completes
both
Review Comment:
`to the buffer size`
##########
modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CollectIntegrationTest.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.integration;
+
+import org.junit.Test;
+
+/**
+ * Integration test for collect node.
+ */
+public class CollectIntegrationTest extends AbstractBasicIntegrationTest {
Review Comment:
One dedicated with-join test only for row in-out and buffer processing.
Based on a plan which can variate. Maybe we need a special node execution test,
WDYT?
##########
modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CollectNode.java:
##########
@@ -114,6 +114,11 @@ public static <Row> CollectNode<Row>
createCountCollector(ExecutionContext<Row>
if (waiting == 0)
source().request(waiting = IN_BUFFER_SIZE);
+ else if (waiting < 0) {
Review Comment:
It look like delayed `downstream().end()`. Why not in own `void end()`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
