This is an automated email from the ASF dual-hosted git repository.
lgbo-ustc pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new cb2c4ca157 [GLUTEN-12358][FLINK] Fix session lookup for parallel
Gluten tasks (#12363)
cb2c4ca157 is described below
commit cb2c4ca157b409dcf1dd551aacd9a19e178c6db2
Author: kevinyhzou <[email protected]>
AuthorDate: Thu Jul 2 09:47:05 2026 +0800
[GLUTEN-12358][FLINK] Fix session lookup for parallel Gluten tasks (#12363)
---
.github/workflows/flink.yml | 2 +-
gluten-flink/docs/Flink.md | 2 +-
.../runtime/operators/GlutenOneInputOperator.java | 5 +-
...onResources.java => GlutenSessionResource.java} | 26 ---
.../runtime/operators/GlutenSourceFunction.java | 3 +-
.../operators/GlutenTaskSessionContext.java | 68 ++++++++
.../runtime/operators/GlutenTwoInputOperator.java | 5 +-
.../typeutils/GlutenStatefulRecordSerializer.java | 27 +--
.../GlutenStatefulRecordSerializerTest.java | 184 +++++++++++++++++++++
gluten-flink/ut/src/test/resources/nexmark/q12.sql | 20 +++
10 files changed, 299 insertions(+), 43 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index d2198a8cfa..656a08d5cf 100644
--- a/.github/workflows/flink.yml
+++ b/.github/workflows/flink.yml
@@ -88,7 +88,7 @@ jobs:
export fmt_SOURCE=BUNDLED
export folly_SOURCE=BUNDLED
git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
- cd velox4j && git reset --hard
97fc1edafebd0f505e613d260f77f92f5252d048
+ cd velox4j && git reset --hard
9811c337d5e37ab8d16d5bf0791b48836fcd2afd
git apply $GITHUB_WORKSPACE/gluten-flink/patches/fix-velox4j.patch
$GITHUB_WORKSPACE/build/mvn clean install -DskipTests -Dgpg.skip
-Dspotless.skip=true
cd ..
diff --git a/gluten-flink/docs/Flink.md b/gluten-flink/docs/Flink.md
index 66387b782d..87f9a4e24b 100644
--- a/gluten-flink/docs/Flink.md
+++ b/gluten-flink/docs/Flink.md
@@ -48,7 +48,7 @@ As some features have not been committed to upstream, you
have to use the follow
## fetch velox4j code
git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
cd velox4j
-git reset --hard 97fc1edafebd0f505e613d260f77f92f5252d048
+git reset --hard 9811c337d5e37ab8d16d5bf0791b48836fcd2afd
mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
```
**Get gluten**
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
index 05645fda9a..09e2b2bb62 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
@@ -120,7 +120,7 @@ public class GlutenOneInputOperator<IN, OUT> extends
TableStreamOperator<OUT>
outputBridge = VectorOutputBridge.Factory.create(outClass);
}
sessionResource = new GlutenSessionResource();
- GlutenSessionResources.getInstance().addSessionResource(id,
sessionResource);
+ GlutenTaskSessionContext.addSessionResource(id, sessionResource);
inputQueue =
sessionResource.getSession().externalStreamOps().newBlockingQueue();
// add a mock input as velox not allow the source is empty.
if (inputType == null) {
@@ -283,6 +283,9 @@ public class GlutenOneInputOperator<IN, OUT> extends
TableStreamOperator<OUT>
inputQueue.close();
}
},
+ () -> {
+ GlutenTaskSessionContext.unregisterSessionResource(id);
+ },
() -> {
if (sessionResource != null) {
sessionResource.close();
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSessionResources.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSessionResource.java
similarity index 76%
rename from
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSessionResources.java
rename to
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSessionResource.java
index cda410f719..ea38229e95 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSessionResources.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSessionResource.java
@@ -26,9 +26,6 @@ import org.apache.flink.runtime.state.KeyedStateBackend;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
-import java.util.HashMap;
-import java.util.Map;
-
// Manage the session and resource for Velox.
class GlutenSessionResource {
private Session session;
@@ -77,26 +74,3 @@ class GlutenSessionResource {
this.keyedStateBackend = keyedStateBackend;
}
}
-
-public class GlutenSessionResources {
- private static final GlutenSessionResources instance = new
GlutenSessionResources();
- private Map<String, GlutenSessionResource> sessionResources = new
HashMap<>();
-
- private GlutenSessionResources() {}
-
- public static GlutenSessionResources getInstance() {
- return instance;
- }
-
- public GlutenSessionResource getSessionResource(String id) {
- return sessionResources.get(id);
- }
-
- public void addSessionResource(String id, GlutenSessionResource
sessionResource) {
- sessionResources.put(id, sessionResource);
- }
-
- public Session getSession(String id) {
- return sessionResources.get(id).getSession();
- }
-}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
index 19c7a69e99..5dfc8447a9 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
@@ -200,6 +200,7 @@ public class GlutenSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
task = null;
}
if (sessionResource != null) {
+ GlutenTaskSessionContext.unregisterSessionResource(id);
sessionResource.close();
sessionResource = null;
}
@@ -241,7 +242,7 @@ public class GlutenSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
}
sessionResource = new GlutenSessionResource();
- GlutenSessionResources.getInstance().addSessionResource(id,
sessionResource);
+ GlutenTaskSessionContext.addSessionResource(id, sessionResource);
Session session = sessionResource.getSession();
query =
new Query(
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTaskSessionContext.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTaskSessionContext.java
new file mode 100644
index 0000000000..035567dd94
--- /dev/null
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTaskSessionContext.java
@@ -0,0 +1,68 @@
+/*
+ * 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.gluten.table.runtime.operators;
+
+import io.github.zhztheplayer.velox4j.session.Session;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Task-thread-local Gluten runtime context used by operators and
serializers. */
+public final class GlutenTaskSessionContext {
+ private static final ThreadLocal<Map<String, GlutenSessionResource>>
SESSION_RESOURCES =
+ ThreadLocal.withInitial(HashMap::new);
+
+ private GlutenTaskSessionContext() {}
+
+ public static GlutenSessionResource getSessionResource(String id) {
+ return SESSION_RESOURCES.get().get(id);
+ }
+
+ public static void addSessionResource(String id, GlutenSessionResource
sessionResource) {
+ SESSION_RESOURCES.get().put(id, sessionResource);
+ }
+
+ public static void registerSessionResource(String operatorId,
GlutenSessionResource resource) {
+ addSessionResource(operatorId, resource);
+ }
+
+ public static void unregisterSessionResource(String operatorId) {
+ Map<String, GlutenSessionResource> resources = SESSION_RESOURCES.get();
+ resources.remove(operatorId);
+ if (resources.isEmpty()) {
+ SESSION_RESOURCES.remove();
+ }
+ }
+
+ public static Session getSession(String operatorId) {
+ Map<String, GlutenSessionResource> resources = SESSION_RESOURCES.get();
+ GlutenSessionResource resource = resources.get(operatorId);
+ if (resource == null) {
+ throw new IllegalStateException(
+ "No Gluten session registered on the current task thread for
operator "
+ + operatorId
+ + ". Registered operators: "
+ + resources.keySet());
+ }
+ Session session = resource.getSession();
+ if (session == null) {
+ throw new IllegalStateException(
+ "Gluten session is already closed for operator " + operatorId);
+ }
+ return session;
+ }
+}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java
index 1251bb084e..7845e22a5b 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java
@@ -258,6 +258,9 @@ public class GlutenTwoInputOperator<IN, OUT> extends
AbstractStreamOperator<OUT>
task.close();
}
},
+ () -> {
+ GlutenTaskSessionContext.unregisterSessionResource(getId());
+ },
() -> {
if (sessionResource != null) {
sessionResource.close();
@@ -324,7 +327,7 @@ public class GlutenTwoInputOperator<IN, OUT> extends
AbstractStreamOperator<OUT>
}
sessionResource = new GlutenSessionResource();
- GlutenSessionResources.getInstance().addSessionResource(getId(),
sessionResource);
+ GlutenTaskSessionContext.addSessionResource(getId(), sessionResource);
leftInputQueue =
sessionResource.getSession().externalStreamOps().newBlockingQueue();
rightInputQueue =
sessionResource.getSession().externalStreamOps().newBlockingQueue();
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/typeutils/GlutenStatefulRecordSerializer.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/typeutils/GlutenStatefulRecordSerializer.java
index 66fdd7bc51..00e8abaab6 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/typeutils/GlutenStatefulRecordSerializer.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/typeutils/GlutenStatefulRecordSerializer.java
@@ -17,7 +17,7 @@
package org.apache.gluten.table.runtime.typeutils;
import org.apache.gluten.streaming.api.operators.GlutenOperator;
-import org.apache.gluten.table.runtime.operators.GlutenSessionResources;
+import org.apache.gluten.table.runtime.operators.GlutenTaskSessionContext;
import io.github.zhztheplayer.velox4j.data.RowVector;
import io.github.zhztheplayer.velox4j.stateful.StatefulRecord;
@@ -37,16 +37,20 @@ import java.io.IOException;
public class GlutenStatefulRecordSerializer extends
TypeSerializer<StatefulRecord> {
private static final long serialVersionUID = 1L;
private final RowType rowType;
- private final GlutenOperator operator;
+ private final String operatorId;
public GlutenStatefulRecordSerializer(RowType rowType, GlutenOperator
operator) {
+ this(rowType, operator.getId());
+ }
+
+ private GlutenStatefulRecordSerializer(RowType rowType, String operatorId) {
this.rowType = rowType;
- this.operator = operator;
+ this.operatorId = operatorId;
}
@Override
public TypeSerializer<StatefulRecord> duplicate() {
- return new GlutenStatefulRecordSerializer(rowType, operator);
+ return new GlutenStatefulRecordSerializer(rowType, operatorId);
}
@Override
@@ -67,12 +71,11 @@ public class GlutenStatefulRecordSerializer extends
TypeSerializer<StatefulRecor
byte[] str = new byte[len];
source.readFully(str);
RowVector rowVector =
- GlutenSessionResources.getInstance()
- .getSession(operator.getId())
+ GlutenTaskSessionContext.getSession(operatorId)
.baseVectorOps()
.deserializeOne(new String(str))
.asRowVector();
- StatefulRecord record = new StatefulRecord(operator.getId(),
rowVector.id(), 0, false, -1);
+ StatefulRecord record = new StatefulRecord(operatorId, rowVector.id(), 0,
false, -1);
record.setRowVector(rowVector);
return record;
}
@@ -130,7 +133,7 @@ public class GlutenStatefulRecordSerializer extends
TypeSerializer<StatefulRecor
@Override
public TypeSerializerSnapshot<StatefulRecord> snapshotConfiguration() {
- return new RowVectorSerializerSnapshot(rowType, operator);
+ return new RowVectorSerializerSnapshot(rowType, operatorId);
}
/** {@link TypeSerializerSnapshot} for Gluten RowVector.. */
@@ -139,16 +142,16 @@ public class GlutenStatefulRecordSerializer extends
TypeSerializer<StatefulRecor
private static final int CURRENT_VERSION = 1;
private RowType rowType;
- private GlutenOperator operator;
+ private String operatorId;
@SuppressWarnings("unused")
public RowVectorSerializerSnapshot() {
// this constructor is used when restoring from a checkpoint/savepoint.
}
- RowVectorSerializerSnapshot(RowType rowType, GlutenOperator operator) {
+ RowVectorSerializerSnapshot(RowType rowType, String operatorId) {
this.rowType = rowType;
- this.operator = operator;
+ this.operatorId = operatorId;
}
@Override
@@ -165,7 +168,7 @@ public class GlutenStatefulRecordSerializer extends
TypeSerializer<StatefulRecor
@Override
public GlutenStatefulRecordSerializer restoreSerializer() {
- return new GlutenStatefulRecordSerializer(rowType, operator);
+ return new GlutenStatefulRecordSerializer(rowType, operatorId);
}
@Override
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenStatefulRecordSerializerTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenStatefulRecordSerializerTest.java
new file mode 100644
index 0000000000..4c38890446
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/operators/GlutenStatefulRecordSerializerTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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.gluten.table.runtime.operators;
+
+import org.apache.gluten.streaming.api.operators.GlutenOperator;
+import org.apache.gluten.table.runtime.stream.common.Velox4jEnvironment;
+import
org.apache.gluten.table.runtime.typeutils.GlutenStatefulRecordSerializer;
+import org.apache.gluten.util.LogicalTypeConverter;
+import org.apache.gluten.vectorized.FlinkRowToVLVectorConvertor;
+
+import io.github.zhztheplayer.velox4j.data.RowVector;
+import io.github.zhztheplayer.velox4j.plan.StatefulPlanNode;
+import io.github.zhztheplayer.velox4j.stateful.StatefulRecord;
+
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class GlutenStatefulRecordSerializerTest {
+ private static final String OPERATOR_ID = "serializer-test-operator";
+
+ private GlutenSessionResource sourceResource;
+ private GlutenSessionResource targetResource;
+
+ @BeforeAll
+ public static void initializeVelox() {
+ Velox4jEnvironment.initializeOnce();
+ }
+
+ @AfterEach
+ public void tearDown() {
+ GlutenTaskSessionContext.unregisterSessionResource(OPERATOR_ID);
+ if (sourceResource != null) {
+ sourceResource.close();
+ sourceResource = null;
+ }
+ if (targetResource != null) {
+ targetResource.close();
+ targetResource = null;
+ }
+ }
+
+ @Test
+ public void testDeserializeUsesTaskLocalSession() throws Exception {
+ io.github.zhztheplayer.velox4j.type.RowType veloxRowType =
createVeloxRowType();
+ GlutenStatefulRecordSerializer serializer =
+ new GlutenStatefulRecordSerializer(veloxRowType, new
TestGlutenOperator(veloxRowType));
+
+ sourceResource = new GlutenSessionResource();
+ RowVector sourceVector =
+ FlinkRowToVLVectorConvertor.fromRowData(
+ GenericRowData.of(7, StringData.fromString("Alice")),
+ sourceResource.getAllocator(),
+ sourceResource.getSession(),
+ veloxRowType);
+ StatefulRecord sourceRecord = new StatefulRecord(OPERATOR_ID,
sourceVector.id(), 0, false, 3);
+ sourceRecord.setRowVector(sourceVector);
+
+ DataOutputSerializer output = new DataOutputSerializer(128);
+ serializer.serialize(sourceRecord, output);
+
+ targetResource = new GlutenSessionResource();
+ GlutenTaskSessionContext.addSessionResource(OPERATOR_ID, targetResource);
+
+ StatefulRecord restoredRecord =
+ serializer.deserialize(new
DataInputDeserializer(output.getCopyOfBuffer()));
+ List<RowData> restoredRows =
+ FlinkRowToVLVectorConvertor.toRowData(
+ restoredRecord.getRowVector(), targetResource.getAllocator(),
veloxRowType);
+
+ assertThat(restoredRecord.getNodeId()).isEqualTo(OPERATOR_ID);
+ assertThat(restoredRows).hasSize(1);
+ RowData restoredRow = restoredRows.get(0);
+ assertThat(restoredRow.getInt(0)).isEqualTo(7);
+
assertThat(restoredRow.getString(1)).isEqualTo(StringData.fromString("Alice"));
+
+ restoredRecord.close();
+ sourceRecord.close();
+ }
+
+ @Test
+ public void testTaskLocalSessionContextIsThreadIsolated() throws Exception {
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ CyclicBarrier barrier = new CyclicBarrier(2);
+ try {
+ Callable<Void> assertion =
+ () -> {
+ assertTaskLocalSessionIsIsolated(barrier);
+ return null;
+ };
+ Future<?> first = executor.submit(assertion);
+ Future<?> second = executor.submit(assertion);
+
+ first.get();
+ second.get();
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ private static void assertTaskLocalSessionIsIsolated(CyclicBarrier barrier)
throws Exception {
+ GlutenSessionResource resource = new GlutenSessionResource();
+ try {
+ GlutenTaskSessionContext.addSessionResource(OPERATOR_ID, resource);
+ barrier.await();
+
+
assertThat(GlutenTaskSessionContext.getSession(OPERATOR_ID)).isSameAs(resource.getSession());
+ } finally {
+ GlutenTaskSessionContext.unregisterSessionResource(OPERATOR_ID);
+ resource.close();
+ }
+ }
+
+ private static io.github.zhztheplayer.velox4j.type.RowType
createVeloxRowType() {
+ RowType flinkRowType =
+ RowType.of(
+ new LogicalType[] {new IntType(), new
VarCharType(VarCharType.MAX_LENGTH)},
+ new String[] {"id", "name"});
+ return (io.github.zhztheplayer.velox4j.type.RowType)
+ LogicalTypeConverter.toVLType(flinkRowType);
+ }
+
+ private static class TestGlutenOperator implements GlutenOperator {
+ private final io.github.zhztheplayer.velox4j.type.RowType rowType;
+
+ private TestGlutenOperator(io.github.zhztheplayer.velox4j.type.RowType
rowType) {
+ this.rowType = rowType;
+ }
+
+ @Override
+ public StatefulPlanNode getPlanNode() {
+ return null;
+ }
+
+ @Override
+ public io.github.zhztheplayer.velox4j.type.RowType getInputType() {
+ return rowType;
+ }
+
+ @Override
+ public Map<String, io.github.zhztheplayer.velox4j.type.RowType>
getOutputTypes() {
+ return Map.of(OPERATOR_ID, rowType);
+ }
+
+ @Override
+ public String getId() {
+ return OPERATOR_ID;
+ }
+ }
+}
diff --git a/gluten-flink/ut/src/test/resources/nexmark/q12.sql
b/gluten-flink/ut/src/test/resources/nexmark/q12.sql
new file mode 100644
index 0000000000..f2cda4f463
--- /dev/null
+++ b/gluten-flink/ut/src/test/resources/nexmark/q12.sql
@@ -0,0 +1,20 @@
+CREATE TABLE nexmark_q12 (
+ bidder BIGINT,
+ bid_count BIGINT,
+ starttime TIMESTAMP(3),
+ endtime TIMESTAMP(3)
+) WITH (
+ 'connector' = 'blackhole'
+);
+
+CREATE VIEW B AS SELECT *, PROCTIME() as p_time FROM bid;
+
+INSERT INTO nexmark_q12
+SELECT
+ bidder,
+ count(*) as bid_count,
+ window_start AS starttime,
+ window_end AS endtime
+FROM TABLE(
+ TUMBLE(TABLE B, DESCRIPTOR(p_time), INTERVAL '10' SECOND))
+GROUP BY bidder, window_start, window_end;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
