You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@geaflow.apache.org by "qingwen220 (via GitHub)" <gi...@apache.org> on 2025/09/16 09:26:27 UTC
[PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
qingwen220 opened a new pull request, #623:
URL: https://github.com/apache/geaflow/pull/623
### What changes were proposed in this pull request?
GraphView type supports Paimon type. e.g.,
```
CREATE GRAPH dy_modern (
Vertex person (
id bigint ID,
name varchar,
age int
),
Edge knows (
srcId bigint SOURCE ID,
targetId bigint DESTINATION ID,
weight double
)
) WITH (
storeType='paimon',
shardCount = 1
);
```
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354227579
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/BasePaimonGraphStore.java:
##########
@@ -19,27 +19,75 @@
package org.apache.geaflow.store.paimon;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_DATABASE;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_EDGE_TABLE;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_INDEX_TABLE;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_VERTEX_TABLE;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.geaflow.common.config.Configuration;
import org.apache.geaflow.state.pushdown.inner.CodeGenFilterConverter;
import org.apache.geaflow.state.pushdown.inner.DirectFilterConverter;
import org.apache.geaflow.state.pushdown.inner.IFilterConverter;
import org.apache.geaflow.store.api.graph.IPushDownStore;
import org.apache.geaflow.store.config.StoreConfigKeys;
import org.apache.geaflow.store.context.StoreContext;
+import org.apache.paimon.catalog.Identifier;
public abstract class BasePaimonGraphStore extends BasePaimonStore implements IPushDownStore {
protected IFilterConverter filterConverter;
+ protected String vertexTable;
+ protected String edgeTable;
+ protected String indexTable;
@Override
public void init(StoreContext storeContext) {
super.init(storeContext);
- boolean codegenEnable =
- storeContext.getConfig().getBoolean(StoreConfigKeys.STORE_FILTER_CODEGEN_ENABLE);
- filterConverter = codegenEnable ? new CodeGenFilterConverter() : new DirectFilterConverter();
+ Configuration config = storeContext.getConfig();
+ boolean codegenEnable = config.getBoolean(StoreConfigKeys.STORE_FILTER_CODEGEN_ENABLE);
+ filterConverter =
+ codegenEnable ? new CodeGenFilterConverter() : new DirectFilterConverter();
+ // use distributed table when distributed mode or database is configured.
+ if (config.contains(PAIMON_STORE_DATABASE) || isDistributedMode) {
+ paimonStoreName = config.getString(PAIMON_STORE_DATABASE);
+ vertexTable = config.getString(PAIMON_STORE_VERTEX_TABLE);
+ edgeTable = config.getString(PAIMON_STORE_EDGE_TABLE);
+ indexTable = config.getString(PAIMON_STORE_INDEX_TABLE);
+ }
}
@Override
public IFilterConverter getFilterConverter() {
return filterConverter;
}
+
+ protected PaimonTableRWHandle createVertexTable(int shardId) {
+ String tableName = vertexTable;
+ if (StringUtils.isEmpty(tableName)) {
+ tableName = String.format("%s#%s", "vertex", shardId);
Review Comment:
Fallback is the current impl, and I currently keep the part unchanged.
The table format string is referenced only here.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "kitalkuyo-gita (via GitHub)" <gi...@apache.org>.
kitalkuyo-gita commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2355154570
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
Review Comment:
OK, thank you for your answer.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "Loognqiang (via GitHub)" <gi...@apache.org>.
Loognqiang merged PR #623:
URL: https://github.com/apache/geaflow/pull/623
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354803197
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
Review Comment:
1. [scenario1] iIf checkpoint 100 is successfully, Geaflow will not rewrite old data.
If checkpoint 100 is partially successful, this will not affect Paimon, because paimon commit is global and the last node of the pipeline, which means it could determine whether the whole checkpoint is successful or not. Only a successful commit could be visible to the reader.
2. [scenario2][scenario3] Do you mean d1 and d2 are out of order between the materialized node and the global commit node?
There are two cases: 1) d1 and d2 are from the same task, the order is guaranteed by the shuffle module. 2) d1 and d2 are from different tasks, and they are written in different buckets. Paimon buckets are independent from each other, because they are in different LSM trees, and compaction happens just in a tree, but not across multiple trees.
3.[scenario4] It wouldn't happen that two tasks write to the same bucket with the current implementation.
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
Review Comment:
1. [scenario1] iIf checkpoint 100 is successfully, Geaflow will not rewrite old data.
If checkpoint 100 is partially successful, this will not affect Paimon, because paimon commit is global and the last node of the pipeline, which means it could determine whether the whole checkpoint is successful or not. Only a successful commit could be visible to the reader.
2. [scenario2][scenario3] Do you mean d1 and d2 are out of order between the materialized node and the global commit node?
There are two cases: 1) d1 and d2 are from the same task, the order is guaranteed by the shuffle module. 2) d1 and d2 are from different tasks, and they are written in different buckets. Paimon buckets are independent from each other, because they are in different LSM trees, and compaction happens just in a tree, but not across multiple trees.
3. [scenario4] It wouldn't happen that two tasks write to the same bucket with the current implementation.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354197904
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/GraphViewMaterializeOp.java:
##########
@@ -105,12 +113,28 @@ public void checkpoint(long windowId) {
this.graphState.manage().operate().setCheckpointId(checkpointId);
this.graphState.manage().operate().finish();
this.graphState.manage().operate().archive();
+
+ if (graphViewDesc.getBackend() == BackendType.Paimon) {
+ int taskIndex = runtimeContext.getTaskArgs().getTaskIndex();
+
+ PaimonCommitRegistry registry = PaimonCommitRegistry.getInstance();
+ List<TaskCommitMessage> messages = registry.pollMessages(taskIndex);
+ if (messages != null && !messages.isEmpty()) {
+ for (TaskCommitMessage message : messages) {
+ List<CommitMessage> msg = message.getMessages();
Review Comment:
empty messages are filtered out before being put into the registry.
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/GraphViewMaterializeOp.java:
##########
@@ -105,12 +113,28 @@ public void checkpoint(long windowId) {
this.graphState.manage().operate().setCheckpointId(checkpointId);
this.graphState.manage().operate().finish();
this.graphState.manage().operate().archive();
+
+ if (graphViewDesc.getBackend() == BackendType.Paimon) {
+ int taskIndex = runtimeContext.getTaskArgs().getTaskIndex();
+
+ PaimonCommitRegistry registry = PaimonCommitRegistry.getInstance();
+ List<TaskCommitMessage> messages = registry.pollMessages(taskIndex);
+ if (messages != null && !messages.isEmpty()) {
+ for (TaskCommitMessage message : messages) {
+ List<CommitMessage> msg = message.getMessages();
+ LOGGER.info("task {} emits windowId:{} chkId:{} table:{} messages:{}",
+ taskIndex, windowId, checkpointId, message.getTableName(),
+ msg.size());
+ collectValue(new PaimonMessage(checkpointId, message.getTableName(),
Review Comment:
no
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "kitalkuyo-gita (via GitHub)" <gi...@apache.org>.
kitalkuyo-gita commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354159515
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/BasePaimonGraphStore.java:
##########
@@ -19,27 +19,75 @@
package org.apache.geaflow.store.paimon;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_DATABASE;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_EDGE_TABLE;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_INDEX_TABLE;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_VERTEX_TABLE;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.geaflow.common.config.Configuration;
import org.apache.geaflow.state.pushdown.inner.CodeGenFilterConverter;
import org.apache.geaflow.state.pushdown.inner.DirectFilterConverter;
import org.apache.geaflow.state.pushdown.inner.IFilterConverter;
import org.apache.geaflow.store.api.graph.IPushDownStore;
import org.apache.geaflow.store.config.StoreConfigKeys;
import org.apache.geaflow.store.context.StoreContext;
+import org.apache.paimon.catalog.Identifier;
public abstract class BasePaimonGraphStore extends BasePaimonStore implements IPushDownStore {
protected IFilterConverter filterConverter;
+ protected String vertexTable;
+ protected String edgeTable;
+ protected String indexTable;
@Override
public void init(StoreContext storeContext) {
super.init(storeContext);
- boolean codegenEnable =
- storeContext.getConfig().getBoolean(StoreConfigKeys.STORE_FILTER_CODEGEN_ENABLE);
- filterConverter = codegenEnable ? new CodeGenFilterConverter() : new DirectFilterConverter();
+ Configuration config = storeContext.getConfig();
+ boolean codegenEnable = config.getBoolean(StoreConfigKeys.STORE_FILTER_CODEGEN_ENABLE);
+ filterConverter =
+ codegenEnable ? new CodeGenFilterConverter() : new DirectFilterConverter();
+ // use distributed table when distributed mode or database is configured.
+ if (config.contains(PAIMON_STORE_DATABASE) || isDistributedMode) {
+ paimonStoreName = config.getString(PAIMON_STORE_DATABASE);
+ vertexTable = config.getString(PAIMON_STORE_VERTEX_TABLE);
+ edgeTable = config.getString(PAIMON_STORE_EDGE_TABLE);
+ indexTable = config.getString(PAIMON_STORE_INDEX_TABLE);
+ }
}
@Override
public IFilterConverter getFilterConverter() {
return filterConverter;
}
+
+ protected PaimonTableRWHandle createVertexTable(int shardId) {
+ String tableName = vertexTable;
+ if (StringUtils.isEmpty(tableName)) {
+ tableName = String.format("%s#%s", "vertex", shardId);
Review Comment:
- When reading the config, if no configuration is found, a fallback is executed. However, is the fallback string consistent with other locations? (For example, is the tableName concatenation format consistent?)
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "kitalkuyo-gita (via GitHub)" <gi...@apache.org>.
kitalkuyo-gita commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2357236722
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/GraphViewMaterializeOp.java:
##########
@@ -105,12 +113,28 @@ public void checkpoint(long windowId) {
this.graphState.manage().operate().setCheckpointId(checkpointId);
this.graphState.manage().operate().finish();
this.graphState.manage().operate().archive();
+
+ if (graphViewDesc.getBackend() == BackendType.Paimon) {
+ int taskIndex = runtimeContext.getTaskArgs().getTaskIndex();
+
+ PaimonCommitRegistry registry = PaimonCommitRegistry.getInstance();
+ List<TaskCommitMessage> messages = registry.pollMessages(taskIndex);
+ if (messages != null && !messages.isEmpty()) {
+ for (TaskCommitMessage message : messages) {
+ List<CommitMessage> msg = message.getMessages();
Review Comment:
ok
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354422608
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
Review Comment:
I don't quite understand what you said. Let's come back to the original comment.
`For multiple messages with the same tableName, there may be merging, duplication, and order requirements, and it is necessary to ensure that they are not lost or confused.`
1. The `message` here is the paimon commit message. Multiple messages come from different materialized tasks with each writing its own bucket. e.g. task1 writes to bucket 1, task2 writes to bucket2, ....
2. These messages are passing to the downstream PaimonGlobalSink (parallelism=1), and do the final commit.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "kitalkuyo-gita (via GitHub)" <gi...@apache.org>.
kitalkuyo-gita commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354154899
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/GraphViewMaterializeOp.java:
##########
@@ -105,12 +113,28 @@ public void checkpoint(long windowId) {
this.graphState.manage().operate().setCheckpointId(checkpointId);
this.graphState.manage().operate().finish();
this.graphState.manage().operate().archive();
+
+ if (graphViewDesc.getBackend() == BackendType.Paimon) {
+ int taskIndex = runtimeContext.getTaskArgs().getTaskIndex();
+
+ PaimonCommitRegistry registry = PaimonCommitRegistry.getInstance();
+ List<TaskCommitMessage> messages = registry.pollMessages(taskIndex);
+ if (messages != null && !messages.isEmpty()) {
+ for (TaskCommitMessage message : messages) {
+ List<CommitMessage> msg = message.getMessages();
+ LOGGER.info("task {} emits windowId:{} chkId:{} table:{} messages:{}",
+ taskIndex, windowId, checkpointId, message.getTableName(),
+ msg.size());
+ collectValue(new PaimonMessage(checkpointId, message.getTableName(),
Review Comment:
Is it allowed to pass an empty list to collectValue? Will it cause downstream errors?
eq
```
List<CommitMessage> msg = message.getMessages();
...
collectValue(new PaimonMessage(checkpointId, message.getTableName(),
msg));
```
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354197904
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/GraphViewMaterializeOp.java:
##########
@@ -105,12 +113,28 @@ public void checkpoint(long windowId) {
this.graphState.manage().operate().setCheckpointId(checkpointId);
this.graphState.manage().operate().finish();
this.graphState.manage().operate().archive();
+
+ if (graphViewDesc.getBackend() == BackendType.Paimon) {
+ int taskIndex = runtimeContext.getTaskArgs().getTaskIndex();
+
+ PaimonCommitRegistry registry = PaimonCommitRegistry.getInstance();
+ List<TaskCommitMessage> messages = registry.pollMessages(taskIndex);
+ if (messages != null && !messages.isEmpty()) {
+ for (TaskCommitMessage message : messages) {
+ List<CommitMessage> msg = message.getMessages();
Review Comment:
no, empty messages will be filtered before being put into the registry.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354231343
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/PaimonTableRWHandle.java:
##########
@@ -63,13 +80,33 @@ public void write(GenericRow row, int bucket) {
}
public void commit(long checkpointId) {
- try (StreamTableCommit commit = table.newStreamWriteBuilder().newCommit()) {
- flush(checkpointId);
- commit.commit(checkpointId, commitMessages);
- commitMessages.clear();
- } catch (Exception e) {
- throw new GeaflowRuntimeException("Failed to commit data into Paimon.", e);
+ commit(checkpointId, false);
+ }
+
+ public void commit(long checkpointId, boolean waitCompaction) {
+ flush(checkpointId, waitCompaction);
+ List<CommitMessage> messages = new ArrayList<>();
+ for (CommitMessage commitMessage : commitMessages) {
+ if (((CommitMessageImpl) commitMessage).isEmpty()) {
Review Comment:
There is only one implementation of CommitMessage. If the cast fails, an exception should be thrown.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354208697
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
Review Comment:
a task writes into its own bucket in the same table, so I think there is no need to de-duplicate and re-order messages from different tasks. The framework will guarantees the order of messages from one task.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354232212
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/commit/PaimonCommitRegistry.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.geaflow.store.paimon.commit;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.paimon.table.sink.CommitMessage;
+
+public class PaimonCommitRegistry {
+
+ private static PaimonCommitRegistry instance;
+
+ private final ConcurrentHashMap<Integer, List<TaskCommitMessage>> index2CommitMessages =
+ new ConcurrentHashMap<>();
+
+ public static synchronized PaimonCommitRegistry initInstance() {
+ if (instance == null) {
+ instance = new PaimonCommitRegistry();
+ }
+ return instance;
+ }
+
+ public static synchronized PaimonCommitRegistry getInstance() {
+ return instance;
+ }
+
+ public synchronized void addMessages(int index, String tableName,
+ List<CommitMessage> commitMessages) {
+ List<TaskCommitMessage> message = index2CommitMessages.computeIfAbsent(index,
+ key -> new ArrayList<>());
+ message.add(new TaskCommitMessage(tableName, commitMessages));
+ }
+
+ public synchronized List<TaskCommitMessage> pollMessages(int index) {
+ List<TaskCommitMessage> messages = index2CommitMessages.get(index);
+ if (messages != null && !messages.isEmpty()) {
+ List<TaskCommitMessage> result = new ArrayList<>(messages);
+ messages.clear();
Review Comment:
The two methods in this class are all synchronized.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354231343
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/PaimonTableRWHandle.java:
##########
@@ -63,13 +80,33 @@ public void write(GenericRow row, int bucket) {
}
public void commit(long checkpointId) {
- try (StreamTableCommit commit = table.newStreamWriteBuilder().newCommit()) {
- flush(checkpointId);
- commit.commit(checkpointId, commitMessages);
- commitMessages.clear();
- } catch (Exception e) {
- throw new GeaflowRuntimeException("Failed to commit data into Paimon.", e);
+ commit(checkpointId, false);
+ }
+
+ public void commit(long checkpointId, boolean waitCompaction) {
+ flush(checkpointId, waitCompaction);
+ List<CommitMessage> messages = new ArrayList<>();
+ for (CommitMessage commitMessage : commitMessages) {
+ if (((CommitMessageImpl) commitMessage).isEmpty()) {
Review Comment:
1. There is only one implementation of CommitMessage.
2. added an instanceof check
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354245196
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
+ List<CommitMessage> commitMessages = tableMessages.computeIfAbsent(tableName,
+ k -> new ArrayList<>());
+ commitMessages.addAll(messages);
+ }
+ }
+ long deserializeTime = System.currentTimeMillis() - startTime;
+ if (!tableMessages.isEmpty()) {
+ for (Map.Entry<String, List<CommitMessage>> entry : tableMessages.entrySet()) {
+ LOGGER.info("commit table:{} messages:{}", entry.getKey(), entry.getValue().size());
+ StreamWriteBuilder writeBuilder = getWriteBuilder(entry.getKey());
+ try (StreamTableCommit commit = writeBuilder.newCommit()) {
+ try {
+ commit.commit(checkpointId, entry.getValue());
+ } catch (Throwable e) {
+ LOGGER.warn("commit failed: {}", e.getMessage(), e);
+ Map<Long, List<CommitMessage>> commitIdAndMessages = new HashMap<>();
+ commitIdAndMessages.put(checkpointId, entry.getValue());
+ commit.filterAndCommit(commitIdAndMessages);
+ }
+ } catch (Throwable e) {
+ LOGGER.error("Failed to commit data into Paimon: {}", e.getMessage(), e);
+ throw new GeaflowRuntimeException("Failed to commit data into Paimon.", e);
+ }
+ }
+ }
+ LOGGER.info("committed chkId:{} messages:{} deserializeCost:{}ms",
+ checkpointId, paimonMessages.size(), deserializeTime);
+ paimonMessages.clear();
+ }
+
+ private StreamWriteBuilder getWriteBuilder(String tableName) {
+ return writeBuilders.computeIfAbsent(tableName, k -> {
+ try {
+ FileStoreTable table = (FileStoreTable) client.getTable(Identifier.create(dbName,
Review Comment:
OK, add the error message in the exception.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "kitalkuyo-gita (via GitHub)" <gi...@apache.org>.
kitalkuyo-gita commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354231111
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
Review Comment:
I'll give you a few examples to help you evaluate them.
#### Scenario 1: Task Retries Cause Data Duplicates
Suppose:
There are two tasks, TaskA (responsible for bucket 0) and TaskB (responsible for bucket 1), both writing to the same table, TableX.
TaskA wrote data D1 at checkpoint 100. Due to network jitter or a node failure, TaskA retried the write.
Data Flow:
```
TaskA first writes D1: {key: "k1", value: "v1"}
TaskA retries due to a failure and writes D1 again: {key: "k1", value: "v1"}
```
Paimon: Two identical entries appear in TableX bucket 0.
Consequences:
If the underlying table does not have a primary key constraint, multiple duplicate entries will result, leading to inaccurate query results and storage bloat.
For example, when performing count/distinct operations, the results may double.
#### Scenario 2: Out-of-order message writing affects eventual consistency
Assume:
```
Task A first writes D1 ({key: "k1", value: "v1"}) and then writes D2 ({key: "k1", value: "v2"}).
Due to distributed network latency, D2 is consumed/written to disk by downstream clients before D1 arrives.
Data flow:
After D2 is written to disk, the value of k1 changes to "v2".
D1 then writes to disk overwriting D2, and the value of k1 returns to "v1".
```
Consequence:
Without global ordering guarantees, the final table state will be inconsistent with the expected state (business-wise, k1 should be v2).
This type of "rollback" caused by out-of-order processing is common in distributed scenarios, especially during failover, recovery, and shuffle operations.
#### Scenario 3: Data Loss or Disorganization After Partition Reassignment or Compaction
Assumptions:
Paimon internally supports bucket compaction, which may merge bucket data from multiple tasks.
TaskA writes to D1, and TaskB writes to D2. During the underlying compaction process, the data in the two buckets is merged.
Data Flow:
```
D1 and D2 are written to their own buckets.
During the compaction process, without proper deduplication and ordering logic, data in D1 and D2 may be duplicated, missing, or even overwritten out of order.
```
Consequences:
When querying TableX, multiple versions of the same key may appear, or the latest data may be lost.
"Eventual consistency" and "idempotence" cannot be guaranteed.
#### Scenario 4: Multiple Tasks Writing to the Same Bucket (Theoretical Conflict)
Assumptions:
After adjustments to certain algorithms (such as hash partitioning), TaskA and TaskB may both write to bucket 0.
Both tasks write the same key, but with different values. Data flow:
```
TaskA writes {key: "k2", value: "A"}
TaskB writes {key: "k2", value: "B"}
```
Finally, bucket 0 contains two entries with the same key but different values, and the order is uncontrollable.
Consequences:
Without global deduplication, ordering, and conflict resolution mechanisms, the final data state depends on the timing of physical writes to disk, which may not meet business expectations.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354197904
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/GraphViewMaterializeOp.java:
##########
@@ -105,12 +113,28 @@ public void checkpoint(long windowId) {
this.graphState.manage().operate().setCheckpointId(checkpointId);
this.graphState.manage().operate().finish();
this.graphState.manage().operate().archive();
+
+ if (graphViewDesc.getBackend() == BackendType.Paimon) {
+ int taskIndex = runtimeContext.getTaskArgs().getTaskIndex();
+
+ PaimonCommitRegistry registry = PaimonCommitRegistry.getInstance();
+ List<TaskCommitMessage> messages = registry.pollMessages(taskIndex);
+ if (messages != null && !messages.isEmpty()) {
+ for (TaskCommitMessage message : messages) {
+ List<CommitMessage> msg = message.getMessages();
Review Comment:
no, if empty message will be filtered before being put into the registry.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "kitalkuyo-gita (via GitHub)" <gi...@apache.org>.
kitalkuyo-gita commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354154678
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/GraphViewMaterializeOp.java:
##########
@@ -105,12 +113,28 @@ public void checkpoint(long windowId) {
this.graphState.manage().operate().setCheckpointId(checkpointId);
this.graphState.manage().operate().finish();
this.graphState.manage().operate().archive();
+
+ if (graphViewDesc.getBackend() == BackendType.Paimon) {
+ int taskIndex = runtimeContext.getTaskArgs().getTaskIndex();
+
+ PaimonCommitRegistry registry = PaimonCommitRegistry.getInstance();
+ List<TaskCommitMessage> messages = registry.pollMessages(taskIndex);
+ if (messages != null && !messages.isEmpty()) {
+ for (TaskCommitMessage message : messages) {
+ List<CommitMessage> msg = message.getMessages();
Review Comment:
message.getMessages() may also return null, so it's recommended to add a null check.
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/GraphViewMaterializeOp.java:
##########
@@ -105,12 +113,28 @@ public void checkpoint(long windowId) {
this.graphState.manage().operate().setCheckpointId(checkpointId);
this.graphState.manage().operate().finish();
this.graphState.manage().operate().archive();
+
+ if (graphViewDesc.getBackend() == BackendType.Paimon) {
+ int taskIndex = runtimeContext.getTaskArgs().getTaskIndex();
+
+ PaimonCommitRegistry registry = PaimonCommitRegistry.getInstance();
+ List<TaskCommitMessage> messages = registry.pollMessages(taskIndex);
+ if (messages != null && !messages.isEmpty()) {
+ for (TaskCommitMessage message : messages) {
+ List<CommitMessage> msg = message.getMessages();
+ LOGGER.info("task {} emits windowId:{} chkId:{} table:{} messages:{}",
+ taskIndex, windowId, checkpointId, message.getTableName(),
+ msg.size());
+ collectValue(new PaimonMessage(checkpointId, message.getTableName(),
Review Comment:
Is it allowed to pass an empty list to collectValue? Will it cause downstream errors?
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
+ List<CommitMessage> commitMessages = tableMessages.computeIfAbsent(tableName,
+ k -> new ArrayList<>());
+ commitMessages.addAll(messages);
+ }
+ }
+ long deserializeTime = System.currentTimeMillis() - startTime;
+ if (!tableMessages.isEmpty()) {
+ for (Map.Entry<String, List<CommitMessage>> entry : tableMessages.entrySet()) {
+ LOGGER.info("commit table:{} messages:{}", entry.getKey(), entry.getValue().size());
+ StreamWriteBuilder writeBuilder = getWriteBuilder(entry.getKey());
+ try (StreamTableCommit commit = writeBuilder.newCommit()) {
+ try {
+ commit.commit(checkpointId, entry.getValue());
+ } catch (Throwable e) {
+ LOGGER.warn("commit failed: {}", e.getMessage(), e);
+ Map<Long, List<CommitMessage>> commitIdAndMessages = new HashMap<>();
+ commitIdAndMessages.put(checkpointId, entry.getValue());
+ commit.filterAndCommit(commitIdAndMessages);
+ }
+ } catch (Throwable e) {
+ LOGGER.error("Failed to commit data into Paimon: {}", e.getMessage(), e);
+ throw new GeaflowRuntimeException("Failed to commit data into Paimon.", e);
+ }
+ }
+ }
+ LOGGER.info("committed chkId:{} messages:{} deserializeCost:{}ms",
+ checkpointId, paimonMessages.size(), deserializeTime);
+ paimonMessages.clear();
+ }
+
+ private StreamWriteBuilder getWriteBuilder(String tableName) {
+ return writeBuilders.computeIfAbsent(tableName, k -> {
+ try {
+ FileStoreTable table = (FileStoreTable) client.getTable(Identifier.create(dbName,
Review Comment:
If client.getTable throws an exception, it will be wrapped and thrown; however, if the table does not exist and auto-create is not enabled, an error will be reported directly. It is recommended to provide clear exception information.
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/BasePaimonGraphStore.java:
##########
@@ -19,27 +19,75 @@
package org.apache.geaflow.store.paimon;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_DATABASE;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_EDGE_TABLE;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_INDEX_TABLE;
+import static org.apache.geaflow.store.paimon.config.PaimonConfigKeys.PAIMON_STORE_VERTEX_TABLE;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.geaflow.common.config.Configuration;
import org.apache.geaflow.state.pushdown.inner.CodeGenFilterConverter;
import org.apache.geaflow.state.pushdown.inner.DirectFilterConverter;
import org.apache.geaflow.state.pushdown.inner.IFilterConverter;
import org.apache.geaflow.store.api.graph.IPushDownStore;
import org.apache.geaflow.store.config.StoreConfigKeys;
import org.apache.geaflow.store.context.StoreContext;
+import org.apache.paimon.catalog.Identifier;
public abstract class BasePaimonGraphStore extends BasePaimonStore implements IPushDownStore {
protected IFilterConverter filterConverter;
+ protected String vertexTable;
+ protected String edgeTable;
+ protected String indexTable;
@Override
public void init(StoreContext storeContext) {
super.init(storeContext);
- boolean codegenEnable =
- storeContext.getConfig().getBoolean(StoreConfigKeys.STORE_FILTER_CODEGEN_ENABLE);
- filterConverter = codegenEnable ? new CodeGenFilterConverter() : new DirectFilterConverter();
+ Configuration config = storeContext.getConfig();
+ boolean codegenEnable = config.getBoolean(StoreConfigKeys.STORE_FILTER_CODEGEN_ENABLE);
+ filterConverter =
+ codegenEnable ? new CodeGenFilterConverter() : new DirectFilterConverter();
+ // use distributed table when distributed mode or database is configured.
+ if (config.contains(PAIMON_STORE_DATABASE) || isDistributedMode) {
+ paimonStoreName = config.getString(PAIMON_STORE_DATABASE);
+ vertexTable = config.getString(PAIMON_STORE_VERTEX_TABLE);
+ edgeTable = config.getString(PAIMON_STORE_EDGE_TABLE);
+ indexTable = config.getString(PAIMON_STORE_INDEX_TABLE);
+ }
}
@Override
public IFilterConverter getFilterConverter() {
return filterConverter;
}
+
+ protected PaimonTableRWHandle createVertexTable(int shardId) {
+ String tableName = vertexTable;
+ if (StringUtils.isEmpty(tableName)) {
+ tableName = String.format("%s#%s", "vertex", shardId);
Review Comment:
- When reading the config, if no configuration is found, a fallback is executed. However, is the fallback string consistent with other locations? (For example, is the tableName concatenation format consistent?)
- StringUtils.isEmpty(edgeTable) However, tableName is vertex, so there may be a copy-paste error. Please verify carefully.
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
Review Comment:
For multiple messages with the same tableName, there may be merging, duplication, and order requirements, and it is necessary to ensure that they are not lost or confused.
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/PaimonTableRWHandle.java:
##########
@@ -63,13 +80,33 @@ public void write(GenericRow row, int bucket) {
}
public void commit(long checkpointId) {
- try (StreamTableCommit commit = table.newStreamWriteBuilder().newCommit()) {
- flush(checkpointId);
- commit.commit(checkpointId, commitMessages);
- commitMessages.clear();
- } catch (Exception e) {
- throw new GeaflowRuntimeException("Failed to commit data into Paimon.", e);
+ commit(checkpointId, false);
+ }
+
+ public void commit(long checkpointId, boolean waitCompaction) {
+ flush(checkpointId, waitCompaction);
+ List<CommitMessage> messages = new ArrayList<>();
+ for (CommitMessage commitMessage : commitMessages) {
+ if (((CommitMessageImpl) commitMessage).isEmpty()) {
Review Comment:
Force the type conversion to CommitMessageImpl. If it's not of this type, an exception will be thrown. Ensure that all CommitMessages implement this implementation, or add an instanceof check.
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/commit/PaimonCommitRegistry.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.geaflow.store.paimon.commit;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.paimon.table.sink.CommitMessage;
+
+public class PaimonCommitRegistry {
+
+ private static PaimonCommitRegistry instance;
+
+ private final ConcurrentHashMap<Integer, List<TaskCommitMessage>> index2CommitMessages =
+ new ConcurrentHashMap<>();
+
+ public static synchronized PaimonCommitRegistry initInstance() {
+ if (instance == null) {
+ instance = new PaimonCommitRegistry();
+ }
+ return instance;
+ }
+
+ public static synchronized PaimonCommitRegistry getInstance() {
+ return instance;
+ }
+
+ public synchronized void addMessages(int index, String tableName,
+ List<CommitMessage> commitMessages) {
+ List<TaskCommitMessage> message = index2CommitMessages.computeIfAbsent(index,
+ key -> new ArrayList<>());
+ message.add(new TaskCommitMessage(tableName, commitMessages));
+ }
+
+ public synchronized List<TaskCommitMessage> pollMessages(int index) {
+ List<TaskCommitMessage> messages = index2CommitMessages.get(index);
+ if (messages != null && !messages.isEmpty()) {
+ List<TaskCommitMessage> result = new ArrayList<>(messages);
+ messages.clear();
Review Comment:
In concurrent scenarios, there may be data consistency risks when other threads read messages after messages.clear() is called.
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/PaimonTableRWHandle.java:
##########
@@ -63,13 +80,33 @@ public void write(GenericRow row, int bucket) {
}
public void commit(long checkpointId) {
- try (StreamTableCommit commit = table.newStreamWriteBuilder().newCommit()) {
- flush(checkpointId);
- commit.commit(checkpointId, commitMessages);
- commitMessages.clear();
- } catch (Exception e) {
- throw new GeaflowRuntimeException("Failed to commit data into Paimon.", e);
+ commit(checkpointId, false);
+ }
+
+ public void commit(long checkpointId, boolean waitCompaction) {
+ flush(checkpointId, waitCompaction);
Review Comment:
Does flush(checkpointId, waitCompaction) ensure that commitMessages is not empty? If so, can the downstream process it?
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "kitalkuyo-gita (via GitHub)" <gi...@apache.org>.
kitalkuyo-gita commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354224614
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
Review Comment:
> 每个任务都会写入同一张表中自己的 bucket，所以我认为没有必要对来自不同任务的消息进行去重和重新排序。框架会保证来自同一个任务的消息的顺序。
1. Although each task is responsible for its own bucket, in a distributed system, multiple tasks may concurrently write to the same table. Especially if the underlying table implementation uses mechanisms such as merging, compression, and repartitioning, message ordering and deduplication cannot rely solely on the local order of a single task.
2. The framework guarantees order within the task itself, but the final state of the global table may be affected by writes from multiple tasks. If messages are duplicated (such as in retry scenarios), the lack of deduplication can lead to data bloat or logical errors.
Regarding your question about the framework guaranteeing deduplication, Paimon explicitly requires or recommends that upstream sinks implement deduplication and ordering guarantees. Otherwise, data anomalies may occur during recovery, replay, and failover.
However, if the downstream table has a primary key constraint, duplicate messages will be discarded or overwritten. If not, data inconsistency may result.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354222492
##########
geaflow/geaflow-plugins/geaflow-store/geaflow-store-paimon/src/main/java/org/apache/geaflow/store/paimon/PaimonTableRWHandle.java:
##########
@@ -63,13 +80,33 @@ public void write(GenericRow row, int bucket) {
}
public void commit(long checkpointId) {
- try (StreamTableCommit commit = table.newStreamWriteBuilder().newCommit()) {
- flush(checkpointId);
- commit.commit(checkpointId, commitMessages);
- commitMessages.clear();
- } catch (Exception e) {
- throw new GeaflowRuntimeException("Failed to commit data into Paimon.", e);
+ commit(checkpointId, false);
+ }
+
+ public void commit(long checkpointId, boolean waitCompaction) {
+ flush(checkpointId, waitCompaction);
Review Comment:
flush does not need to ensure that. It just do prepare commit.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org
Re: [PR] [ISSUE-622] feat: support distributed writing with one Paimon graph store [geaflow]
Posted by "qingwen220 (via GitHub)" <gi...@apache.org>.
qingwen220 commented on code in PR #623:
URL: https://github.com/apache/geaflow/pull/623#discussion_r2354422608
##########
geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/materialize/PaimonGlobalSink.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.geaflow.operator.impl.graph.materialize;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.api.function.RichWindowFunction;
+import org.apache.geaflow.api.function.io.SinkFunction;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.store.paimon.PaimonCatalogClient;
+import org.apache.geaflow.store.paimon.PaimonCatalogManager;
+import org.apache.geaflow.store.paimon.commit.PaimonMessage;
+import org.apache.geaflow.store.paimon.config.PaimonConfigKeys;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PaimonGlobalSink extends RichWindowFunction implements SinkFunction<PaimonMessage> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PaimonGlobalSink.class);
+
+ private String dbName;
+ private String jobName;
+ private long windowId;
+
+ private List<PaimonMessage> paimonMessages;
+ private Map<String, StreamWriteBuilder> writeBuilders;
+ private PaimonCatalogClient client;
+
+ @Override
+ public void open(RuntimeContext runtimeContext) {
+ this.windowId = runtimeContext.getWindowId();
+ Configuration jobConfig = runtimeContext.getConfiguration();
+ this.client = PaimonCatalogManager.getCatalogClient(jobConfig);
+ this.jobName = jobConfig.getString(ExecutionConfigKeys.JOB_APP_NAME);
+ this.dbName = jobConfig.getString(PaimonConfigKeys.PAIMON_STORE_DATABASE);
+
+ this.writeBuilders = new HashMap<>();
+ this.paimonMessages = new ArrayList<>();
+ LOGGER.info("init paimon sink with db {}", dbName);
+ }
+
+ @Override
+ public void write(PaimonMessage message) throws Exception {
+ paimonMessages.add(message);
+ }
+
+ @Override
+ public void finish() {
+ if (paimonMessages.isEmpty()) {
+ LOGGER.info("commit windowId {} empty messages", windowId);
+ return;
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long checkpointId = paimonMessages.get(0).getCheckpointId();
+ Map<String, List<CommitMessage>> tableMessages = new HashMap<>();
+ for (PaimonMessage message : paimonMessages) {
+ List<CommitMessage> messages = message.getMessages();
+ if (messages != null && !messages.isEmpty()) {
+ String tableName = message.getTableName();
Review Comment:
I don't quite understand what you said. Let's come back to the original comment.
`For multiple messages with the same tableName, there may be merging, duplication, and order requirements, and it is necessary to ensure that they are not lost or confused.`
1. The `message` here is the paimon commit message. Multiple messages come from different materialized tasks with each writing its own bucket. e.g. task1 writes to bucket 1, task2 writes to bucket2, ....
2. These messages are forwarded to the downstream PaimonGlobalSink (parallelism=1), and do the final commit.
--
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: issues-unsubscribe@geaflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@geaflow.apache.org
For additional commands, e-mail: issues-help@geaflow.apache.org