You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/10/07 10:40:38 UTC
[GitHub] [beam] AdalbertMemSQL opened a new pull request, #23535: SingleStoreIO
AdalbertMemSQL opened a new pull request, #23535:
URL: https://github.com/apache/beam/pull/23535
Implemented SingleStoreIO according to this design doc:
https://docs.google.com/document/d/1WU-hkoZ93SaGXyOz_UtX0jXzIRl194hCId_IdmEV9jw/edit?usp=sharing
addresses #22617
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/get-started-contributing/#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1317447575
Succeeded: https://ci-beam.apache.org/view/PerformanceTests/job/beam_PerformanceTests_SingleStoreIO/9/console
metrics:
```
13:10:34 org.apache.beam.sdk.io.singlestore.SingleStoreIOIT > testWriteThenRead STANDARD_OUT
13:10:34 Load test results for test (ID): fef7d33b-660b-4c77-a441-22fd9d6d548a and timestamp: 2022-11-16T18:10:34.119000000Z:
13:10:34 Metric: Value:
13:10:34 write_time 25.008
13:10:34 Load test results for test (ID): fef7d33b-660b-4c77-a441-22fd9d6d548a and timestamp: 2022-11-16T18:10:34.119000000Z:
13:10:34 Metric: Value:
13:10:34 read_time 74.979
13:10:34 Load test results for test (ID): fef7d33b-660b-4c77-a441-22fd9d6d548a and timestamp: 2022-11-16T18:10:34.119000000Z:
13:10:34 Metric: Value:
13:10:34 read_with_partitions_time 28.704
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314528771
Notice that SingleStoreIO Performance Test actually runs on direct runner. Direct runner is only for local debug use. It adds some redundant calculations for validations and is not performant. Could you please consider running it on Dataflow runner (there is an `switches("-DintegrationTestRunner=dataflow")` could refer to other performance tests)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314040939
Hi @AdalbertMemSQL would you mind rebase the branch onto latest master and I can run a seedjob (make beam_PerformanceTests_SingleStoreIO and Java_SingleStore_IO_Direct work), thanks
also please consider adding an updates to https://github.com/apache/beam/blob/master/CHANGES.md
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1292136199
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @kileys for label java.
R: @Abacn for label build.
R: @pabloem for label io.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn merged pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn merged PR #23535:
URL: https://github.com/apache/beam/pull/23535
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1296880627
retest this please
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1296897402
Run CommunityMetrics PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] johnjcasey commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
johnjcasey commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1016754087
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/Read.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Filter;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle;
+import org.apache.beam.sdk.transforms.SerializableFunctions;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@AutoValue
+public abstract class Read<T> extends PTransform<PBegin, PCollection<T>> {
Review Comment:
some class level commentary would be appreciated, even if it just points to documentation in SingleStoreIO
##########
sdks/java/io/singlestore/build.gradle:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(
+ automaticModuleName: 'org.apache.beam.sdk.io.singlestore',
+)
+provideIntegrationTestingDependencies()
+enableJavaPerformanceTesting()
+
+description = "Apache Beam :: SDKs :: Java :: IO :: SingleStore"
+ext.summary = "IO to read and write on SingleStore datasource."
+
+dependencies {
+ implementation library.java.vendored_guava_26_0_jre
+ implementation library.java.joda_time
+ implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation group: 'com.singlestore', name: 'singlestore-jdbc-client', version: '1.1.4'
+ implementation library.java.slf4j_api
+ implementation "org.apache.commons:commons-dbcp2:2.8.0"
Review Comment:
can we move these dependencies to BeamModulePlugin.groovy?
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/ReadWithPartitions.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@AutoValue
+public abstract class ReadWithPartitions<T> extends PTransform<PBegin, PCollection<T>> {
Review Comment:
because this is public, some class level commentary would be good
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/ReadWithPartitions.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@AutoValue
+public abstract class ReadWithPartitions<T> extends PTransform<PBegin, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(ReadWithPartitions.class);
+
+ abstract @Nullable DataSourceConfiguration getDataSourceConfiguration();
+
+ abstract @Nullable ValueProvider<String> getQuery();
+
+ abstract @Nullable ValueProvider<String> getTable();
+
+ abstract @Nullable RowMapper<T> getRowMapper();
+
+ abstract @Nullable ValueProvider<Integer> getInitialNumReaders();
+
+ abstract Builder<T> toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder<T> {
+ abstract Builder<T> setDataSourceConfiguration(DataSourceConfiguration dataSourceConfiguration);
+
+ abstract Builder<T> setQuery(ValueProvider<String> query);
+
+ abstract Builder<T> setTable(ValueProvider<String> table);
+
+ abstract Builder<T> setRowMapper(RowMapper<T> rowMapper);
+
+ abstract Builder<T> setInitialNumReaders(ValueProvider<Integer> initialNumReaders);
+
+ abstract ReadWithPartitions<T> build();
+ }
+
+ public ReadWithPartitions<T> withDataSourceConfiguration(DataSourceConfiguration config) {
+ checkNotNull(config, "dataSourceConfiguration can not be null");
+ return toBuilder().setDataSourceConfiguration(config).build();
+ }
+
+ public ReadWithPartitions<T> withQuery(String query) {
+ checkNotNull(query, "query can not be null");
+ return withQuery(ValueProvider.StaticValueProvider.of(query));
+ }
+
+ public ReadWithPartitions<T> withQuery(ValueProvider<String> query) {
+ checkNotNull(query, "query can not be null");
+ return toBuilder().setQuery(query).build();
+ }
+
+ public ReadWithPartitions<T> withTable(String table) {
+ checkNotNull(table, "table can not be null");
+ return withTable(ValueProvider.StaticValueProvider.of(table));
+ }
+
+ public ReadWithPartitions<T> withTable(ValueProvider<String> table) {
+ checkNotNull(table, "table can not be null");
+ return toBuilder().setTable(table).build();
+ }
+
+ public ReadWithPartitions<T> withRowMapper(RowMapper<T> rowMapper) {
+ checkNotNull(rowMapper, "rowMapper can not be null");
+ return toBuilder().setRowMapper(rowMapper).build();
+ }
+
+ /** Pre-split initial restriction and start initialNumReaders reading at the very beginning. */
+ public ReadWithPartitions<T> withInitialNumReaders(Integer initialNumReaders) {
+ checkNotNull(initialNumReaders, "initialNumReaders can not be null");
+ return withInitialNumReaders(ValueProvider.StaticValueProvider.of(initialNumReaders));
+ }
+
+ /** Same as {@link #withInitialNumReaders(Integer)} but accepting a ValueProvider. */
+ public ReadWithPartitions<T> withInitialNumReaders(ValueProvider<Integer> initialNumReaders) {
+ checkNotNull(initialNumReaders, "initialNumReaders can not be null");
+ return toBuilder().setInitialNumReaders(initialNumReaders).build();
+ }
+
+ @Override
+ public PCollection<T> expand(PBegin input) {
+ DataSourceConfiguration dataSourceConfiguration =
+ Util.getRequiredArgument(
+ getDataSourceConfiguration(), "withDataSourceConfiguration() is required");
+ String database =
+ Util.getRequiredArgument(
+ dataSourceConfiguration.getDatabase(),
+ "withDatabase() is required for DataSourceConfiguration in order to perform readWithPartitions");
+ RowMapper<T> rowMapper =
+ Util.getRequiredArgument(getRowMapper(), "withRowMapper() is required");
+
+ int initialNumReaders = Util.getArgumentWithDefault(getInitialNumReaders(), 1);
+ checkArgument(
+ initialNumReaders >= 1, "withInitialNumReaders() should be greater or equal to 1");
+
+ String actualQuery = Util.getSelectQuery(getTable(), getQuery());
+
+ Coder<T> coder =
+ Util.inferCoder(
+ rowMapper,
+ input.getPipeline().getCoderRegistry(),
+ input.getPipeline().getSchemaRegistry(),
+ LOG);
+
+ return input
+ .apply(Create.of((Void) null))
+ .apply(
+ ParDo.of(
+ new ReadWithPartitions.ReadWithPartitionsFn<>(
+ dataSourceConfiguration, actualQuery, database, rowMapper, initialNumReaders)))
+ .setCoder(coder);
+ }
+
+ private static class ReadWithPartitionsFn<ParameterT, OutputT> extends DoFn<ParameterT, OutputT> {
+ DataSourceConfiguration dataSourceConfiguration;
+ String query;
+ String database;
+ RowMapper<OutputT> rowMapper;
+ int initialNumReaders;
+
+ ReadWithPartitionsFn(
+ DataSourceConfiguration dataSourceConfiguration,
+ String query,
+ String database,
+ RowMapper<OutputT> rowMapper,
+ int initialNumReaders) {
+ this.dataSourceConfiguration = dataSourceConfiguration;
+ this.query = query;
+ this.database = database;
+ this.rowMapper = rowMapper;
+ this.initialNumReaders = initialNumReaders;
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext context, RestrictionTracker<OffsetRange, Long> tracker) throws Exception {
+ DataSource dataSource = dataSourceConfiguration.getDataSource();
+ Connection conn = dataSource.getConnection();
+ try {
+ for (long partition = tracker.currentRestriction().getFrom();
Review Comment:
this should probably not have partition++. In principle, this will always cause an error, because at some point the tryClaim will attempt to claim an invalid partition.
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/Write.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VoidCoder;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+
+@AutoValue
+public abstract class Write<T> extends PTransform<PCollection<T>, PDone> {
Review Comment:
Class level commentary please
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314505151
Run Java_Kafka_IO_Direct PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314810698
@Abacn Can you now run the perf test?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314444320
Run Java SingleStoreIO Performance Test
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314033654
Run Seed Job
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314484382
integration test on direct runner not running because there is no pipeline options specified so it is skipped, I guess. Could feed with a small data size (in contrast to the performance test which uses a larger data size "100000").
Performance test now failing with
```
17:04:21 org.apache.beam.sdk.io.singlestore.SingleStoreIOIT > testWriteThenRead FAILED
17:04:21 java.sql.SQLNonTransientConnectionException: Socket fail to connect to host:address=(host=LOAD_BALANCER_IP)(port=3306)(type=primary). LOAD_BALANCER_IP
```
Anyway currently there is only 1 integration test + performance test. We can just make the performance test work at this moment (temporarily disable SingleStore_IO_Direct groovy script). For integration test running on direct runner docker environment needed to be set in `@BeforeClass` and persumably some helper class are needed. Could be a followup.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314488982
I just fixed this error in the performance test.
Looking into the integration one.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1017862110
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/Write.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VoidCoder;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+
+@AutoValue
+public abstract class Write<T> extends PTransform<PCollection<T>, PDone> {
+
+ private static final int DEFAULT_BATCH_SIZE = 100000;
+ private static final int BUFFER_SIZE = 524288;
+
+ abstract @Nullable DataSourceConfiguration getDataSourceConfiguration();
+
+ abstract @Nullable ValueProvider<String> getTable();
+
+ abstract @Nullable ValueProvider<Integer> getBatchSize();
+
+ abstract @Nullable UserDataMapper<T> getUserDataMapper();
+
+ abstract Builder<T> toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder<T> {
+ abstract Builder<T> setDataSourceConfiguration(DataSourceConfiguration dataSourceConfiguration);
+
+ abstract Builder<T> setTable(ValueProvider<String> table);
+
+ abstract Builder<T> setBatchSize(ValueProvider<Integer> batchSize);
+
+ abstract Builder<T> setUserDataMapper(UserDataMapper<T> userDataMapper);
+
+ abstract Write<T> build();
+ }
+
+ public Write<T> withDataSourceConfiguration(DataSourceConfiguration config) {
+ checkNotNull(config, "dataSourceConfiguration can not be null");
+ return toBuilder().setDataSourceConfiguration(config).build();
+ }
+
+ public Write<T> withTable(String table) {
+ checkNotNull(table, "table can not be null");
+ return withTable(ValueProvider.StaticValueProvider.of(table));
+ }
+
+ public Write<T> withTable(ValueProvider<String> table) {
+ checkNotNull(table, "table can not be null");
+ return toBuilder().setTable(table).build();
+ }
+
+ public Write<T> withUserDataMapper(UserDataMapper<T> userDataMapper) {
+ checkNotNull(userDataMapper, "userDataMapper can not be null");
+ return toBuilder().setUserDataMapper(userDataMapper).build();
+ }
+
+ /**
+ * Provide a maximum number of rows that is written by one SQL statement. Default is 100000.
+ *
+ * @param batchSize maximum number of rows that is written by one SQL statement
+ */
+ public Write<T> withBatchSize(Integer batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return withBatchSize(ValueProvider.StaticValueProvider.of(batchSize));
+ }
+
+ /** Same as {@link #withBatchSize(Integer)} but accepting a ValueProvider. */
+ public Write<T> withBatchSize(ValueProvider<Integer> batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return toBuilder().setBatchSize(batchSize).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<T> input) {
+ DataSourceConfiguration dataSourceConfiguration =
+ Util.getRequiredArgument(
+ getDataSourceConfiguration(), "withDataSourceConfiguration() is required");
+ String table = Util.getRequiredArgument(getTable(), "withTable() is required");
+ UserDataMapper<T> userDataMapper =
+ Util.getRequiredArgument(getUserDataMapper(), "withUserDataMapper() is required");
+ int batchSize = Util.getArgumentWithDefault(getBatchSize(), DEFAULT_BATCH_SIZE);
+ checkArgument(batchSize > 0, "batchSize should be greater then 0");
+
+ input
+ .apply(
+ ParDo.of(
+ new DoFn<T, List<String>>() {
+ @ProcessElement
+ public void processElement(ProcessContext context) throws Exception {
+ context.output(userDataMapper.mapRow(context.element()));
+ }
+ }))
+ .setCoder(ListCoder.of(StringUtf8Coder.of()))
+ .apply(ParDo.of(new BatchFn<>(batchSize)))
+ .apply(ParDo.of(new WriteFn<Void>(dataSourceConfiguration, table)))
+ .setCoder(VoidCoder.of());
+
+ return PDone.in(input.getPipeline());
+ }
+
+ private static class BatchFn<ParameterT> extends DoFn<ParameterT, Iterable<ParameterT>> {
+ List<ParameterT> batch = new ArrayList<>();
+ int batchSize;
+
+ BatchFn(int batchSize) {
+ this.batchSize = batchSize;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ batch.add(context.element());
+ if (batch.size() >= batchSize) {
+ context.output(batch);
+ batch = new ArrayList<>();
+ }
+ }
+
+ @FinishBundle
+ public void finish(FinishBundleContext context) {
+ if (batch.size() > 0) {
+ context.output(batch, Instant.now(), GlobalWindow.INSTANCE);
+ batch = new ArrayList<>();
+ }
+ }
+ }
+
+ private static class WriteFn<OutputT> extends DoFn<Iterable<List<String>>, OutputT> {
+ DataSourceConfiguration dataSourceConfiguration;
+ String table;
+
+ WriteFn(DataSourceConfiguration dataSourceConfiguration, String table) {
+ this.dataSourceConfiguration = dataSourceConfiguration;
+ this.table = table;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) throws Exception {
+ DataSource dataSource = dataSourceConfiguration.getDataSource();
+
+ Connection conn = dataSource.getConnection();
+ try {
+ Statement stmt = conn.createStatement();
+ try (PipedOutputStream baseStream = new PipedOutputStream();
+ InputStream inputStream = new PipedInputStream(baseStream, BUFFER_SIZE)) {
+ ((com.singlestore.jdbc.Statement) ((DelegatingStatement) stmt).getInnermostDelegate())
+ .setNextLocalInfileInputStream(inputStream);
+
+ final Exception[] writeException = new Exception[1];
+
+ Thread dataWritingThread =
Review Comment:
Hmm....
I don't know how to implement this asynchronous task using Beam framework.
Here we have two parts - statement execution and writing data to the stream.
They are tightly coupled (writing will get stuck if the statement is not executed and the buffer is full and the statement execution won't finish until writing is finished).
And also these parts share a common PipedOutputStream which is not serializable.
Does the Beam framework have features that will allow something like this?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1018033307
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/Write.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VoidCoder;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+
+@AutoValue
+public abstract class Write<T> extends PTransform<PCollection<T>, PDone> {
+
+ private static final int DEFAULT_BATCH_SIZE = 100000;
+ private static final int BUFFER_SIZE = 524288;
+
+ abstract @Nullable DataSourceConfiguration getDataSourceConfiguration();
+
+ abstract @Nullable ValueProvider<String> getTable();
+
+ abstract @Nullable ValueProvider<Integer> getBatchSize();
+
+ abstract @Nullable UserDataMapper<T> getUserDataMapper();
+
+ abstract Builder<T> toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder<T> {
+ abstract Builder<T> setDataSourceConfiguration(DataSourceConfiguration dataSourceConfiguration);
+
+ abstract Builder<T> setTable(ValueProvider<String> table);
+
+ abstract Builder<T> setBatchSize(ValueProvider<Integer> batchSize);
+
+ abstract Builder<T> setUserDataMapper(UserDataMapper<T> userDataMapper);
+
+ abstract Write<T> build();
+ }
+
+ public Write<T> withDataSourceConfiguration(DataSourceConfiguration config) {
+ checkNotNull(config, "dataSourceConfiguration can not be null");
+ return toBuilder().setDataSourceConfiguration(config).build();
+ }
+
+ public Write<T> withTable(String table) {
+ checkNotNull(table, "table can not be null");
+ return withTable(ValueProvider.StaticValueProvider.of(table));
+ }
+
+ public Write<T> withTable(ValueProvider<String> table) {
+ checkNotNull(table, "table can not be null");
+ return toBuilder().setTable(table).build();
+ }
+
+ public Write<T> withUserDataMapper(UserDataMapper<T> userDataMapper) {
+ checkNotNull(userDataMapper, "userDataMapper can not be null");
+ return toBuilder().setUserDataMapper(userDataMapper).build();
+ }
+
+ /**
+ * Provide a maximum number of rows that is written by one SQL statement. Default is 100000.
+ *
+ * @param batchSize maximum number of rows that is written by one SQL statement
+ */
+ public Write<T> withBatchSize(Integer batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return withBatchSize(ValueProvider.StaticValueProvider.of(batchSize));
+ }
+
+ /** Same as {@link #withBatchSize(Integer)} but accepting a ValueProvider. */
+ public Write<T> withBatchSize(ValueProvider<Integer> batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return toBuilder().setBatchSize(batchSize).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<T> input) {
+ DataSourceConfiguration dataSourceConfiguration =
+ Util.getRequiredArgument(
+ getDataSourceConfiguration(), "withDataSourceConfiguration() is required");
+ String table = Util.getRequiredArgument(getTable(), "withTable() is required");
+ UserDataMapper<T> userDataMapper =
+ Util.getRequiredArgument(getUserDataMapper(), "withUserDataMapper() is required");
+ int batchSize = Util.getArgumentWithDefault(getBatchSize(), DEFAULT_BATCH_SIZE);
+ checkArgument(batchSize > 0, "batchSize should be greater then 0");
+
+ input
+ .apply(
+ ParDo.of(
+ new DoFn<T, List<String>>() {
+ @ProcessElement
+ public void processElement(ProcessContext context) throws Exception {
+ context.output(userDataMapper.mapRow(context.element()));
+ }
+ }))
+ .setCoder(ListCoder.of(StringUtf8Coder.of()))
+ .apply(ParDo.of(new BatchFn<>(batchSize)))
+ .apply(ParDo.of(new WriteFn<Void>(dataSourceConfiguration, table)))
Review Comment:
That makes sense.
I will return a PCollection of integers, where each integer will represent a number of rows written by each LOAD DATA query. It will allow you to check the total number of written rows and see how the task is parallelized.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1017947911
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/ReadWithPartitions.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@AutoValue
+public abstract class ReadWithPartitions<T> extends PTransform<PBegin, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(ReadWithPartitions.class);
+
+ abstract @Nullable DataSourceConfiguration getDataSourceConfiguration();
+
+ abstract @Nullable ValueProvider<String> getQuery();
+
+ abstract @Nullable ValueProvider<String> getTable();
+
+ abstract @Nullable RowMapper<T> getRowMapper();
+
+ abstract @Nullable ValueProvider<Integer> getInitialNumReaders();
+
+ abstract Builder<T> toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder<T> {
+ abstract Builder<T> setDataSourceConfiguration(DataSourceConfiguration dataSourceConfiguration);
+
+ abstract Builder<T> setQuery(ValueProvider<String> query);
+
+ abstract Builder<T> setTable(ValueProvider<String> table);
+
+ abstract Builder<T> setRowMapper(RowMapper<T> rowMapper);
+
+ abstract Builder<T> setInitialNumReaders(ValueProvider<Integer> initialNumReaders);
+
+ abstract ReadWithPartitions<T> build();
+ }
+
+ public ReadWithPartitions<T> withDataSourceConfiguration(DataSourceConfiguration config) {
+ checkNotNull(config, "dataSourceConfiguration can not be null");
+ return toBuilder().setDataSourceConfiguration(config).build();
+ }
+
+ public ReadWithPartitions<T> withQuery(String query) {
+ checkNotNull(query, "query can not be null");
+ return withQuery(ValueProvider.StaticValueProvider.of(query));
+ }
+
+ public ReadWithPartitions<T> withQuery(ValueProvider<String> query) {
+ checkNotNull(query, "query can not be null");
+ return toBuilder().setQuery(query).build();
+ }
+
+ public ReadWithPartitions<T> withTable(String table) {
+ checkNotNull(table, "table can not be null");
+ return withTable(ValueProvider.StaticValueProvider.of(table));
+ }
+
+ public ReadWithPartitions<T> withTable(ValueProvider<String> table) {
+ checkNotNull(table, "table can not be null");
+ return toBuilder().setTable(table).build();
+ }
+
+ public ReadWithPartitions<T> withRowMapper(RowMapper<T> rowMapper) {
+ checkNotNull(rowMapper, "rowMapper can not be null");
+ return toBuilder().setRowMapper(rowMapper).build();
+ }
+
+ /** Pre-split initial restriction and start initialNumReaders reading at the very beginning. */
+ public ReadWithPartitions<T> withInitialNumReaders(Integer initialNumReaders) {
+ checkNotNull(initialNumReaders, "initialNumReaders can not be null");
+ return withInitialNumReaders(ValueProvider.StaticValueProvider.of(initialNumReaders));
+ }
+
+ /** Same as {@link #withInitialNumReaders(Integer)} but accepting a ValueProvider. */
+ public ReadWithPartitions<T> withInitialNumReaders(ValueProvider<Integer> initialNumReaders) {
+ checkNotNull(initialNumReaders, "initialNumReaders can not be null");
+ return toBuilder().setInitialNumReaders(initialNumReaders).build();
+ }
+
+ @Override
+ public PCollection<T> expand(PBegin input) {
+ DataSourceConfiguration dataSourceConfiguration =
+ Util.getRequiredArgument(
+ getDataSourceConfiguration(), "withDataSourceConfiguration() is required");
+ String database =
+ Util.getRequiredArgument(
+ dataSourceConfiguration.getDatabase(),
+ "withDatabase() is required for DataSourceConfiguration in order to perform readWithPartitions");
+ RowMapper<T> rowMapper =
+ Util.getRequiredArgument(getRowMapper(), "withRowMapper() is required");
+
+ int initialNumReaders = Util.getArgumentWithDefault(getInitialNumReaders(), 1);
+ checkArgument(
+ initialNumReaders >= 1, "withInitialNumReaders() should be greater or equal to 1");
+
+ String actualQuery = Util.getSelectQuery(getTable(), getQuery());
+
+ Coder<T> coder =
+ Util.inferCoder(
+ rowMapper,
+ input.getPipeline().getCoderRegistry(),
+ input.getPipeline().getSchemaRegistry(),
+ LOG);
+
+ return input
+ .apply(Create.of((Void) null))
+ .apply(
+ ParDo.of(
+ new ReadWithPartitions.ReadWithPartitionsFn<>(
+ dataSourceConfiguration, actualQuery, database, rowMapper, initialNumReaders)))
+ .setCoder(coder);
+ }
+
+ private static class ReadWithPartitionsFn<ParameterT, OutputT> extends DoFn<ParameterT, OutputT> {
+ DataSourceConfiguration dataSourceConfiguration;
+ String query;
+ String database;
+ RowMapper<OutputT> rowMapper;
+ int initialNumReaders;
+
+ ReadWithPartitionsFn(
+ DataSourceConfiguration dataSourceConfiguration,
+ String query,
+ String database,
+ RowMapper<OutputT> rowMapper,
+ int initialNumReaders) {
+ this.dataSourceConfiguration = dataSourceConfiguration;
+ this.query = query;
+ this.database = database;
+ this.rowMapper = rowMapper;
+ this.initialNumReaders = initialNumReaders;
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext context, RestrictionTracker<OffsetRange, Long> tracker) throws Exception {
+ DataSource dataSource = dataSourceConfiguration.getDataSource();
+ Connection conn = dataSource.getConnection();
+ try {
+ for (long partition = tracker.currentRestriction().getFrom();
Review Comment:
The same loop is presented in this blog https://beam.apache.org/blog/splittable-do-fn/
```
public void process(ProcessContext c, OffsetRangeTracker tracker) {
for (long i = tracker.currentRestriction().getFrom(); tracker.tryClaim(i); ++i) {
c.output(KV.of(c.element().getKey(), i));
}
}
```
I thought that `tryClaim` will just return `false` when an invalid partition is provided.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] johnjcasey commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
johnjcasey commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1008335684
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/DataSourceConfiguration.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.Serializable;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.commons.dbcp2.BasicDataSource;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * A POJO describing a SingleStoreDB {@link DataSource} by providing all properties needed to create
+ * it.
+ */
+@AutoValue
+public abstract class DataSourceConfiguration implements Serializable {
+ abstract @Nullable ValueProvider<String> getEndpoint();
Review Comment:
These don't need to be valueproviders. ValueProvider was needed for legacy templates, but with flextemplates they are no longer required
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/DataSourceConfiguration.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.Serializable;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.commons.dbcp2.BasicDataSource;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * A POJO describing a SingleStoreDB {@link DataSource} by providing all properties needed to create
+ * it.
+ */
+@AutoValue
+public abstract class DataSourceConfiguration implements Serializable {
+ abstract @Nullable ValueProvider<String> getEndpoint();
+
+ abstract @Nullable ValueProvider<String> getUsername();
+
+ abstract @Nullable ValueProvider<String> getPassword();
+
+ abstract @Nullable ValueProvider<String> getDatabase();
+
+ abstract @Nullable ValueProvider<String> getConnectionProperties();
+
+ abstract Builder builder();
+
+ @AutoValue.Builder
+ abstract static class Builder {
+ abstract Builder setEndpoint(ValueProvider<String> endpoint);
+
+ abstract Builder setUsername(ValueProvider<String> username);
+
+ abstract Builder setPassword(ValueProvider<String> password);
+
+ abstract Builder setDatabase(ValueProvider<String> database);
+
+ abstract Builder setConnectionProperties(ValueProvider<String> connectionProperties);
+
+ abstract DataSourceConfiguration build();
+ }
+
+ public static DataSourceConfiguration create(String endpoint) {
+ checkNotNull(endpoint, "endpoint can not be null");
+ return create(ValueProvider.StaticValueProvider.of(endpoint));
+ }
+
+ public static DataSourceConfiguration create(ValueProvider<String> endpoint) {
+ checkNotNull(endpoint, "endpoint can not be null");
+ return new AutoValue_DataSourceConfiguration.Builder().setEndpoint(endpoint).build();
+ }
+
+ public DataSourceConfiguration withUsername(String username) {
+ checkNotNull(username, "username can not be null");
+ return withUsername(ValueProvider.StaticValueProvider.of(username));
+ }
+
+ public DataSourceConfiguration withUsername(ValueProvider<String> username) {
+ checkNotNull(username, "username can not be null");
+ return builder().setUsername(username).build();
+ }
+
+ public DataSourceConfiguration withPassword(String password) {
+ checkNotNull(password, "password can not be null");
+ return withPassword(ValueProvider.StaticValueProvider.of(password));
+ }
+
+ public DataSourceConfiguration withPassword(ValueProvider<String> password) {
+ checkNotNull(password, "password can not be null");
+ return builder().setPassword(password).build();
+ }
+
+ public DataSourceConfiguration withDatabase(String database) {
+ checkNotNull(database, "database can not be null");
+ return withDatabase(ValueProvider.StaticValueProvider.of(database));
+ }
+
+ public DataSourceConfiguration withDatabase(ValueProvider<String> database) {
+ checkNotNull(database, "database can not be null");
+ return builder().setDatabase(database).build();
+ }
+
+ /**
+ * Sets the connection properties passed to driver.connect(...). Format of the string must be
+ * [propertyName=property;]*
+ *
+ * <p>NOTE - The "user" and "password" properties can be add via {@link #withUsername(String)},
+ * {@link #withPassword(String)}, so they do not need to be included here.
+ *
+ * <p>Full list of supported properties can be found here {@link <a
+ * href="https://docs.singlestore.com/managed-service/en/developer-resources/connect-with-application-development-tools/connect-with-java-jdbc/the-singlestore-jdbc-driver.html#connection-string-parameters">...</a>}
+ */
+ public DataSourceConfiguration withConnectionProperties(String connectionProperties) {
+ checkNotNull(connectionProperties, "connectionProperties can not be null");
+ return withConnectionProperties(ValueProvider.StaticValueProvider.of(connectionProperties));
+ }
+
+ /** Same as {@link #withConnectionProperties(String)} but accepting a ValueProvider. */
+ public DataSourceConfiguration withConnectionProperties(
+ ValueProvider<String> connectionProperties) {
+ checkNotNull(connectionProperties, "connectionProperties can not be null");
+ return builder().setConnectionProperties(connectionProperties).build();
+ }
+
+ public static void populateDisplayData(
+ @Nullable DataSourceConfiguration dataSourceConfiguration, DisplayData.Builder builder) {
+ if (dataSourceConfiguration != null) {
+ builder.addIfNotNull(DisplayData.item("endpoint", dataSourceConfiguration.getEndpoint()));
+ builder.addIfNotNull(DisplayData.item("username", dataSourceConfiguration.getUsername()));
+ builder.addIfNotNull(DisplayData.item("database", dataSourceConfiguration.getDatabase()));
+ builder.addIfNotNull(
+ DisplayData.item(
+ "connectionProperties", dataSourceConfiguration.getConnectionProperties()));
+ }
+ }
+
+ public DataSource getDataSource() {
+ String endpoint = Util.getRequiredArgument(getEndpoint(), "endpoint can not be null");
Review Comment:
Prefer using org.apache.beam.sdk.util.Preconditions for some of these checks
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/Util.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import org.apache.beam.sdk.coders.CannotProvideCoderException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderRegistry;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.schemas.NoSuchSchemaException;
+import org.apache.beam.sdk.schemas.SchemaRegistry;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+
+public class Util {
Review Comment:
Can we rename this to SingleStoreUtil, or something more descriptive? Also, please add some class level commentary
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/Write.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VoidCoder;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+
+@AutoValue
+public abstract class Write<T> extends PTransform<PCollection<T>, PDone> {
+
+ private static final int DEFAULT_BATCH_SIZE = 100000;
+ private static final int BUFFER_SIZE = 524288;
+
+ abstract @Nullable DataSourceConfiguration getDataSourceConfiguration();
+
+ abstract @Nullable ValueProvider<String> getTable();
+
+ abstract @Nullable ValueProvider<Integer> getBatchSize();
+
+ abstract @Nullable UserDataMapper<T> getUserDataMapper();
+
+ abstract Builder<T> toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder<T> {
+ abstract Builder<T> setDataSourceConfiguration(DataSourceConfiguration dataSourceConfiguration);
+
+ abstract Builder<T> setTable(ValueProvider<String> table);
+
+ abstract Builder<T> setBatchSize(ValueProvider<Integer> batchSize);
+
+ abstract Builder<T> setUserDataMapper(UserDataMapper<T> userDataMapper);
+
+ abstract Write<T> build();
+ }
+
+ public Write<T> withDataSourceConfiguration(DataSourceConfiguration config) {
+ checkNotNull(config, "dataSourceConfiguration can not be null");
+ return toBuilder().setDataSourceConfiguration(config).build();
+ }
+
+ public Write<T> withTable(String table) {
+ checkNotNull(table, "table can not be null");
+ return withTable(ValueProvider.StaticValueProvider.of(table));
+ }
+
+ public Write<T> withTable(ValueProvider<String> table) {
+ checkNotNull(table, "table can not be null");
+ return toBuilder().setTable(table).build();
+ }
+
+ public Write<T> withUserDataMapper(UserDataMapper<T> userDataMapper) {
+ checkNotNull(userDataMapper, "userDataMapper can not be null");
+ return toBuilder().setUserDataMapper(userDataMapper).build();
+ }
+
+ /**
+ * Provide a maximum number of rows that is written by one SQL statement. Default is 100000.
+ *
+ * @param batchSize maximum number of rows that is written by one SQL statement
+ */
+ public Write<T> withBatchSize(Integer batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return withBatchSize(ValueProvider.StaticValueProvider.of(batchSize));
+ }
+
+ /** Same as {@link #withBatchSize(Integer)} but accepting a ValueProvider. */
+ public Write<T> withBatchSize(ValueProvider<Integer> batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return toBuilder().setBatchSize(batchSize).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<T> input) {
+ DataSourceConfiguration dataSourceConfiguration =
+ Util.getRequiredArgument(
+ getDataSourceConfiguration(), "withDataSourceConfiguration() is required");
+ String table = Util.getRequiredArgument(getTable(), "withTable() is required");
+ UserDataMapper<T> userDataMapper =
+ Util.getRequiredArgument(getUserDataMapper(), "withUserDataMapper() is required");
+ int batchSize = Util.getArgumentWithDefault(getBatchSize(), DEFAULT_BATCH_SIZE);
+ checkArgument(batchSize > 0, "batchSize should be greater then 0");
+
+ input
+ .apply(
+ ParDo.of(
+ new DoFn<T, List<String>>() {
+ @ProcessElement
+ public void processElement(ProcessContext context) throws Exception {
+ context.output(userDataMapper.mapRow(context.element()));
+ }
+ }))
+ .setCoder(ListCoder.of(StringUtf8Coder.of()))
+ .apply(ParDo.of(new BatchFn<>(batchSize)))
+ .apply(ParDo.of(new WriteFn<Void>(dataSourceConfiguration, table)))
Review Comment:
returning void or pdone is considered an antipattern. Better would be to return the status of the writes. This enables users to use the output of this transform to drive future behavior
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/ReadWithPartitions.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@AutoValue
+public abstract class ReadWithPartitions<T> extends PTransform<PBegin, PCollection<T>> {
Review Comment:
Probably worth adding documentation arguing that ReadWithPartitions is preferred for performance reasons
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/ReadWithPartitions.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@AutoValue
+public abstract class ReadWithPartitions<T> extends PTransform<PBegin, PCollection<T>> {
+ private static final Logger LOG = LoggerFactory.getLogger(ReadWithPartitions.class);
+
+ abstract @Nullable DataSourceConfiguration getDataSourceConfiguration();
+
+ abstract @Nullable ValueProvider<String> getQuery();
+
+ abstract @Nullable ValueProvider<String> getTable();
+
+ abstract @Nullable RowMapper<T> getRowMapper();
+
+ abstract @Nullable ValueProvider<Integer> getInitialNumReaders();
+
+ abstract Builder<T> toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder<T> {
+ abstract Builder<T> setDataSourceConfiguration(DataSourceConfiguration dataSourceConfiguration);
+
+ abstract Builder<T> setQuery(ValueProvider<String> query);
+
+ abstract Builder<T> setTable(ValueProvider<String> table);
+
+ abstract Builder<T> setRowMapper(RowMapper<T> rowMapper);
+
+ abstract Builder<T> setInitialNumReaders(ValueProvider<Integer> initialNumReaders);
+
+ abstract ReadWithPartitions<T> build();
+ }
+
+ public ReadWithPartitions<T> withDataSourceConfiguration(DataSourceConfiguration config) {
+ checkNotNull(config, "dataSourceConfiguration can not be null");
+ return toBuilder().setDataSourceConfiguration(config).build();
+ }
+
+ public ReadWithPartitions<T> withQuery(String query) {
+ checkNotNull(query, "query can not be null");
+ return withQuery(ValueProvider.StaticValueProvider.of(query));
+ }
+
+ public ReadWithPartitions<T> withQuery(ValueProvider<String> query) {
+ checkNotNull(query, "query can not be null");
+ return toBuilder().setQuery(query).build();
+ }
+
+ public ReadWithPartitions<T> withTable(String table) {
+ checkNotNull(table, "table can not be null");
+ return withTable(ValueProvider.StaticValueProvider.of(table));
+ }
+
+ public ReadWithPartitions<T> withTable(ValueProvider<String> table) {
+ checkNotNull(table, "table can not be null");
+ return toBuilder().setTable(table).build();
+ }
+
+ public ReadWithPartitions<T> withRowMapper(RowMapper<T> rowMapper) {
+ checkNotNull(rowMapper, "rowMapper can not be null");
+ return toBuilder().setRowMapper(rowMapper).build();
+ }
+
+ /** Pre-split initial restriction and start initialNumReaders reading at the very beginning. */
+ public ReadWithPartitions<T> withInitialNumReaders(Integer initialNumReaders) {
+ checkNotNull(initialNumReaders, "initialNumReaders can not be null");
+ return withInitialNumReaders(ValueProvider.StaticValueProvider.of(initialNumReaders));
+ }
+
+ /** Same as {@link #withInitialNumReaders(Integer)} but accepting a ValueProvider. */
+ public ReadWithPartitions<T> withInitialNumReaders(ValueProvider<Integer> initialNumReaders) {
+ checkNotNull(initialNumReaders, "initialNumReaders can not be null");
+ return toBuilder().setInitialNumReaders(initialNumReaders).build();
+ }
+
+ @Override
+ public PCollection<T> expand(PBegin input) {
+ DataSourceConfiguration dataSourceConfiguration =
+ Util.getRequiredArgument(
+ getDataSourceConfiguration(), "withDataSourceConfiguration() is required");
+ String database =
+ Util.getRequiredArgument(
+ dataSourceConfiguration.getDatabase(),
+ "withDatabase() is required for DataSourceConfiguration in order to perform readWithPartitions");
+ RowMapper<T> rowMapper =
+ Util.getRequiredArgument(getRowMapper(), "withRowMapper() is required");
+
+ int initialNumReaders = Util.getArgumentWithDefault(getInitialNumReaders(), 1);
+ checkArgument(
+ initialNumReaders >= 1, "withInitialNumReaders() should be greater or equal to 1");
+
+ String actualQuery = Util.getSelectQuery(getTable(), getQuery());
+
+ Coder<T> coder =
+ Util.inferCoder(
+ rowMapper,
+ input.getPipeline().getCoderRegistry(),
+ input.getPipeline().getSchemaRegistry(),
+ LOG);
+
+ return input
+ .apply(Create.of((Void) null))
+ .apply(
+ ParDo.of(
+ new ReadWithPartitions.ReadWithPartitionsFn<>(
+ dataSourceConfiguration, actualQuery, database, rowMapper, initialNumReaders)))
+ .setCoder(coder);
+ }
+
+ private static class ReadWithPartitionsFn<ParameterT, OutputT> extends DoFn<ParameterT, OutputT> {
+ DataSourceConfiguration dataSourceConfiguration;
+ String query;
+ String database;
+ RowMapper<OutputT> rowMapper;
+ int initialNumReaders;
+
+ ReadWithPartitionsFn(
+ DataSourceConfiguration dataSourceConfiguration,
+ String query,
+ String database,
+ RowMapper<OutputT> rowMapper,
+ int initialNumReaders) {
+ this.dataSourceConfiguration = dataSourceConfiguration;
+ this.query = query;
+ this.database = database;
+ this.rowMapper = rowMapper;
+ this.initialNumReaders = initialNumReaders;
+ }
+
+ @ProcessElement
+ public void processElement(
+ ProcessContext context, RestrictionTracker<OffsetRange, Long> tracker) throws Exception {
+ DataSource dataSource = dataSourceConfiguration.getDataSource();
+ Connection conn = dataSource.getConnection();
+ try {
+ for (long partition = tracker.currentRestriction().getFrom();
Review Comment:
either because it is outside of the tracker range, or because it is beyond the total partition count.
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/Write.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VoidCoder;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+
+@AutoValue
+public abstract class Write<T> extends PTransform<PCollection<T>, PDone> {
+
+ private static final int DEFAULT_BATCH_SIZE = 100000;
+ private static final int BUFFER_SIZE = 524288;
+
+ abstract @Nullable DataSourceConfiguration getDataSourceConfiguration();
+
+ abstract @Nullable ValueProvider<String> getTable();
+
+ abstract @Nullable ValueProvider<Integer> getBatchSize();
+
+ abstract @Nullable UserDataMapper<T> getUserDataMapper();
+
+ abstract Builder<T> toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder<T> {
+ abstract Builder<T> setDataSourceConfiguration(DataSourceConfiguration dataSourceConfiguration);
+
+ abstract Builder<T> setTable(ValueProvider<String> table);
+
+ abstract Builder<T> setBatchSize(ValueProvider<Integer> batchSize);
+
+ abstract Builder<T> setUserDataMapper(UserDataMapper<T> userDataMapper);
+
+ abstract Write<T> build();
+ }
+
+ public Write<T> withDataSourceConfiguration(DataSourceConfiguration config) {
+ checkNotNull(config, "dataSourceConfiguration can not be null");
+ return toBuilder().setDataSourceConfiguration(config).build();
+ }
+
+ public Write<T> withTable(String table) {
+ checkNotNull(table, "table can not be null");
+ return withTable(ValueProvider.StaticValueProvider.of(table));
+ }
+
+ public Write<T> withTable(ValueProvider<String> table) {
+ checkNotNull(table, "table can not be null");
+ return toBuilder().setTable(table).build();
+ }
+
+ public Write<T> withUserDataMapper(UserDataMapper<T> userDataMapper) {
+ checkNotNull(userDataMapper, "userDataMapper can not be null");
+ return toBuilder().setUserDataMapper(userDataMapper).build();
+ }
+
+ /**
+ * Provide a maximum number of rows that is written by one SQL statement. Default is 100000.
+ *
+ * @param batchSize maximum number of rows that is written by one SQL statement
+ */
+ public Write<T> withBatchSize(Integer batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return withBatchSize(ValueProvider.StaticValueProvider.of(batchSize));
+ }
+
+ /** Same as {@link #withBatchSize(Integer)} but accepting a ValueProvider. */
+ public Write<T> withBatchSize(ValueProvider<Integer> batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return toBuilder().setBatchSize(batchSize).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<T> input) {
+ DataSourceConfiguration dataSourceConfiguration =
+ Util.getRequiredArgument(
+ getDataSourceConfiguration(), "withDataSourceConfiguration() is required");
+ String table = Util.getRequiredArgument(getTable(), "withTable() is required");
+ UserDataMapper<T> userDataMapper =
+ Util.getRequiredArgument(getUserDataMapper(), "withUserDataMapper() is required");
+ int batchSize = Util.getArgumentWithDefault(getBatchSize(), DEFAULT_BATCH_SIZE);
+ checkArgument(batchSize > 0, "batchSize should be greater then 0");
+
+ input
+ .apply(
+ ParDo.of(
+ new DoFn<T, List<String>>() {
+ @ProcessElement
+ public void processElement(ProcessContext context) throws Exception {
+ context.output(userDataMapper.mapRow(context.element()));
+ }
+ }))
+ .setCoder(ListCoder.of(StringUtf8Coder.of()))
+ .apply(ParDo.of(new BatchFn<>(batchSize)))
+ .apply(ParDo.of(new WriteFn<Void>(dataSourceConfiguration, table)))
+ .setCoder(VoidCoder.of());
+
+ return PDone.in(input.getPipeline());
+ }
+
+ private static class BatchFn<ParameterT> extends DoFn<ParameterT, Iterable<ParameterT>> {
+ List<ParameterT> batch = new ArrayList<>();
+ int batchSize;
+
+ BatchFn(int batchSize) {
+ this.batchSize = batchSize;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ batch.add(context.element());
+ if (batch.size() >= batchSize) {
+ context.output(batch);
+ batch = new ArrayList<>();
+ }
+ }
+
+ @FinishBundle
+ public void finish(FinishBundleContext context) {
+ if (batch.size() > 0) {
+ context.output(batch, Instant.now(), GlobalWindow.INSTANCE);
+ batch = new ArrayList<>();
+ }
+ }
+ }
+
+ private static class WriteFn<OutputT> extends DoFn<Iterable<List<String>>, OutputT> {
+ DataSourceConfiguration dataSourceConfiguration;
+ String table;
+
+ WriteFn(DataSourceConfiguration dataSourceConfiguration, String table) {
+ this.dataSourceConfiguration = dataSourceConfiguration;
+ this.table = table;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) throws Exception {
+ DataSource dataSource = dataSourceConfiguration.getDataSource();
+
+ Connection conn = dataSource.getConnection();
+ try {
+ Statement stmt = conn.createStatement();
+ try (PipedOutputStream baseStream = new PipedOutputStream();
+ InputStream inputStream = new PipedInputStream(baseStream, BUFFER_SIZE)) {
+ ((com.singlestore.jdbc.Statement) ((DelegatingStatement) stmt).getInnermostDelegate())
+ .setNextLocalInfileInputStream(inputStream);
+
+ final Exception[] writeException = new Exception[1];
+
+ Thread dataWritingThread =
Review Comment:
Avoid creating threads inside of dofns. In Beam, asynchronous and multithreading behavior is best handled by the Beam framework itself.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1024352346
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/SingleStoreIO.java:
##########
@@ -0,0 +1,955 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Filter;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle;
+import org.apache.beam.sdk.transforms.SerializableFunctions;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.util.Preconditions;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.commons.dbcp2.BasicDataSource;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * IO to read and write data on SingleStoreDB.
+ *
+ * <h3>Reading from SingleStoreDB datasource</h3>
+ *
+ * <p>SingleStoreIO source returns a bounded collection of {@code T} as a {@code PCollection<T>}. T
+ * is the type returned by the provided {@link RowMapper}.
+ *
+ * <p>To configure the SingleStoreDB source, you have to provide a {@link DataSourceConfiguration}
+ * using {@link DataSourceConfiguration#create(String)}(endpoint). Optionally, {@link
+ * DataSourceConfiguration#withUsername(String)} and {@link
+ * DataSourceConfiguration#withPassword(String)} allows you to define username and password.
+ *
+ * <p>For example:
+ *
+ * <pre>{@code
+ * pipeline.apply(SingleStoreIO.<KV<Integer, String>>read()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
+ * .withUsername("username")
+ * .withPassword("password"))
+ * .withQuery("select id, name from Person")
+ * .withRowMapper(new RowMapper<KV<Integer, String>>() {
+ * public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
+ * return KV.of(resultSet.getInt(1), resultSet.getString(2));
+ * }
+ * })
+ * );
+ * }</pre>
+ *
+ * <p>Query parameters can be configured using a user-provided {@link StatementPreparator}. For
+ * example:
+ *
+ * <pre>{@code
+ * pipeline.apply(SingleStoreIO.<KV<Integer, String>>read()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306"))
+ * .withQuery("select id,name from Person where name = ?")
+ * .withStatementPreparator(new StatementPreparator() {
+ * public void setParameters(PreparedStatement preparedStatement) throws Exception {
+ * preparedStatement.setString(1, "Darwin");
+ * }
+ * })
+ * .withRowMapper(new RowMapper<KV<Integer, String>>() {
+ * public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
+ * return KV.of(resultSet.getInt(1), resultSet.getString(2));
+ * }
+ * })
+ * );
+ * }</pre>
+ *
+ * <h4>Parallel reading from a SingleStoreDB datasource</h4>
+ *
+ * <p>SingleStoreIO supports partitioned reading of all data from a table. To enable this, use
+ * {@link SingleStoreIO#readWithPartitions()}. This way of data reading is preferred because of the
+ * performance reasons.
+ *
+ * <p>The following example shows usage of {@link SingleStoreIO#readWithPartitions()}
+ *
+ * <pre>{@code
+ * pipeline.apply(SingleStoreIO.<Row>readWithPartitions()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
+ * .withUsername("username")
+ * .withPassword("password"))
+ * .withTable("Person")
+ * .withRowMapper(new RowMapper<KV<Integer, String>>() {
+ * public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
+ * return KV.of(resultSet.getInt(1), resultSet.getString(2));
+ * }
+ * })
+ * );
+ * }</pre>
+ *
+ * <h3>Writing to SingleStoreDB datasource</h3>
+ *
+ * <p>SingleStoreIO supports writing records into a database. It writes a {@link PCollection} to the
+ * database by converting data to CSV and sending it to the database with LOAD DATA query.
+ *
+ * <p>Like the source, to configure the sink, you have to provide a {@link DataSourceConfiguration}.
+ *
+ * <pre>{@code
+ * pipeline
+ * .apply(...)
+ * .apply(SingleStoreIO.<KV<Integer, String>>write()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
+ * .withUsername("username")
+ * .withPassword("password"))
+ * .withStatement("insert into Person values(?, ?)")
+ * .withUserDataMapper(new UserDataMapper<KV<Integer, String>>() {
+ * @Override
+ * public List<String> mapRow(KV<Integer, String> element) {
+ * List<String> res = new ArrayList<>();
+ * res.add(element.getKey().toString());
+ * res.add(element.getValue());
+ * return res;
+ * }
+ * })
+ * );
+ * }</pre>
+ */
+final class SingleStoreIO {
Review Comment:
Oops I believe this is a mistake. should be `final class SingleStoreUtil` and `public class SingleStoreIO`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1023364390
##########
sdks/java/io/singlestore/src/test/java/org/apache/beam/sdk/io/singlestore/SingleStoreIOIT.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.sdk.io.common.IOITHelper.readIOTestPipelineOptions;
+
+import com.google.cloud.Timestamp;
+import com.singlestore.jdbc.SingleStoreDataSource;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.function.Function;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.io.common.DatabaseTestHelper;
+import org.apache.beam.sdk.io.common.HashingFn;
+import org.apache.beam.sdk.io.common.TestRow;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testutils.NamedTestResult;
+import org.apache.beam.sdk.testutils.metrics.IOITMetrics;
+import org.apache.beam.sdk.testutils.metrics.MetricsReader;
+import org.apache.beam.sdk.testutils.metrics.TimeMonitor;
+import org.apache.beam.sdk.testutils.publishing.InfluxDBSettings;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Sum;
+import org.apache.beam.sdk.transforms.Top;
+import org.apache.beam.sdk.values.PCollection;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class SingleStoreIOIT {
+
+ private static final String NAMESPACE = SingleStoreIOIT.class.getName();
+
+ private static final String DATABASE_NAME = "SingleStoreIOIT";
+
+ private static int numberOfRows;
+
+ private static String tableName;
+
+ private static String serverName;
+
+ private static String username;
+
+ private static String password;
+
+ private static Integer port;
+
+ private static SingleStoreIO.DataSourceConfiguration dataSourceConfiguration;
+
+ private static InfluxDBSettings settings;
+
+ @BeforeClass
+ public static void setup() {
+ SingleStoreIOTestPipelineOptions options;
+ try {
+ options = readIOTestPipelineOptions(SingleStoreIOTestPipelineOptions.class);
+ } catch (IllegalArgumentException e) {
+ options = null;
+ }
+ org.junit.Assume.assumeNotNull(options);
+
+ numberOfRows = options.getNumberOfRecords();
+ serverName = options.getSingleStoreServerName();
+ username = options.getSingleStoreUsername();
+ password = options.getSingleStorePassword();
+ port = options.getSingleStorePort();
+ tableName = DatabaseTestHelper.getTestTableName("IT");
+ dataSourceConfiguration =
+ SingleStoreIO.DataSourceConfiguration.create(serverName + ":" + port)
+ .withDatabase(DATABASE_NAME)
+ .withPassword(password)
+ .withUsername(username);
+ settings =
+ InfluxDBSettings.builder()
+ .withHost(options.getInfluxHost())
+ .withDatabase(options.getInfluxDatabase())
+ .withMeasurement(options.getInfluxMeasurement())
+ .get();
+ }
+
+ void createDatabaseIfNotExists() throws SQLException {
+ DataSource dataSource =
+ new SingleStoreDataSource(
+ String.format(
+ "jdbc:singlestore://%s:%d/?user=%s&password=%s&allowLocalInfile=TRUE",
+ serverName, port, username, password));
+ try (Connection conn = dataSource.getConnection();
+ Statement stmt = conn.createStatement()) {
+ stmt.executeQuery(String.format("CREATE DATABASE IF NOT EXISTS %s", DATABASE_NAME));
+ }
+ }
+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testWriteThenRead() throws Exception {
+ createDatabaseIfNotExists();
+ DataSource dataSource =
+ new SingleStoreDataSource(
+ String.format(
+ "jdbc:singlestore://%s:%d/%s?user=%s&password=%s&allowLocalInfile=TRUE",
+ serverName, port, DATABASE_NAME, username, password));
+ DatabaseTestHelper.createTable(dataSource, tableName);
+ try {
+ PipelineResult writeResult = runWrite();
+ writeResult.waitUntilFinish();
+ PipelineResult readResult = runRead();
+ readResult.waitUntilFinish();
Review Comment:
We should assert the pipeline status for write and read is DONE; we may also assert the number of record we read is consistent with the number of record we have written (some basic validation). Looks like currently if the pipeline fails remotely the test will still pass.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1023358897
##########
.test-infra/jenkins/job_PerformanceTests_SingleStoreIO.groovy:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as common
+import Kubernetes
+import InfluxDBCredentialsHelper
+
+String jobName = "beam_PerformanceTests_SingleStoreIO"
+
+void waitForPodWithLabel(job, Kubernetes k8s, String label) {
+ job.steps {
+ shell("${k8s.KUBERNETES_DIR}/singlestore/wait-for-pod-with-label.sh ${label} 600")
+ }
+}
+
+void waitFor(job, Kubernetes k8s, String resource) {
+ job.steps {
+ shell("${k8s.KUBERNETES_DIR}/singlestore/wait-for.sh ${resource} 600")
+ }
+}
+
+job(jobName) {
+ common.setTopLevelMainJobProperties(delegate)
+ common.setAutoJob(delegate,'H H/6 * * *')
+ common.enablePhraseTriggeringFromPullRequest(
+ delegate,
+ 'Java SingleStoreIO Performance Test',
+ 'Run Java SingleStoreIO Performance Test')
+ InfluxDBCredentialsHelper.useCredentials(delegate)
+
+
+ String namespace = common.getKubernetesNamespace(jobName)
+ String kubeconfigPath = common.getKubeconfigLocationForNamespace(namespace)
+ Kubernetes k8s = Kubernetes.create(delegate, kubeconfigPath, namespace)
+
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-rbac.yaml"))
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-cluster-crd.yaml"))
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-operator.yaml"))
+ waitForPodWithLabel(delegate, k8s, "sdb-operator")
+
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-cluster.yaml"))
+ waitFor(delegate, k8s, "memsqlclusters.memsql.com")
+
+ String singlestoreHostName = "LOAD_BALANCER_IP"
+ k8s.loadBalancerIP("svc-sdb-cluster-ddl", singlestoreHostName)
+
+ Map pipelineOptions = [
+ tempRoot : 'gs://temp-storage-for-perf-tests',
+ project : 'apache-beam-testing',
+ runner : 'DataflowRunner',
+ singleStoreServerName : "\$${singlestoreHostName}",
+ singleStoreUsername : "admin",
+ singleStorePassword : "secretpass",
+ singleStorePort: "3306",
+ numberOfRecords: "100000",
Review Comment:
Sorry I meant 10M records (100 times of "100000").
The metrics shown in https://ci-beam.apache.org/view/PerformanceTests/job/beam_PerformanceTests_SingleStoreIO/6/console
```
13:54:32 Load test results for test (ID): 403b18ab-b3dc-475a-b4d9-0a77866159c8 and timestamp: 2022-11-15T18:54:32.477000000Z:
13:54:32 Metric: Value:
13:54:32 write_time 0.628
13:54:32 Load test results for test (ID): 403b18ab-b3dc-475a-b4d9-0a77866159c8 and timestamp: 2022-11-15T18:54:32.477000000Z:
13:54:33 Metric: Value:
13:54:33 read_time 4.433
13:54:33 Load test results for test (ID): 403b18ab-b3dc-475a-b4d9-0a77866159c8 and timestamp: 2022-11-15T18:54:32.477000000Z:
13:54:33 Metric: Value:
13:54:33 read_with_partitions_time 1.87
```
If performance is linear 1M record still costs seconds to execute write.
SingleStoreDB is pretty performant!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314451251
Looks like the Performance test failed because of wrong script permissions. I changed them.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] johnjcasey commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
johnjcasey commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1018069678
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/Write.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VoidCoder;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+
+@AutoValue
+public abstract class Write<T> extends PTransform<PCollection<T>, PDone> {
+
+ private static final int DEFAULT_BATCH_SIZE = 100000;
+ private static final int BUFFER_SIZE = 524288;
+
+ abstract @Nullable DataSourceConfiguration getDataSourceConfiguration();
+
+ abstract @Nullable ValueProvider<String> getTable();
+
+ abstract @Nullable ValueProvider<Integer> getBatchSize();
+
+ abstract @Nullable UserDataMapper<T> getUserDataMapper();
+
+ abstract Builder<T> toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder<T> {
+ abstract Builder<T> setDataSourceConfiguration(DataSourceConfiguration dataSourceConfiguration);
+
+ abstract Builder<T> setTable(ValueProvider<String> table);
+
+ abstract Builder<T> setBatchSize(ValueProvider<Integer> batchSize);
+
+ abstract Builder<T> setUserDataMapper(UserDataMapper<T> userDataMapper);
+
+ abstract Write<T> build();
+ }
+
+ public Write<T> withDataSourceConfiguration(DataSourceConfiguration config) {
+ checkNotNull(config, "dataSourceConfiguration can not be null");
+ return toBuilder().setDataSourceConfiguration(config).build();
+ }
+
+ public Write<T> withTable(String table) {
+ checkNotNull(table, "table can not be null");
+ return withTable(ValueProvider.StaticValueProvider.of(table));
+ }
+
+ public Write<T> withTable(ValueProvider<String> table) {
+ checkNotNull(table, "table can not be null");
+ return toBuilder().setTable(table).build();
+ }
+
+ public Write<T> withUserDataMapper(UserDataMapper<T> userDataMapper) {
+ checkNotNull(userDataMapper, "userDataMapper can not be null");
+ return toBuilder().setUserDataMapper(userDataMapper).build();
+ }
+
+ /**
+ * Provide a maximum number of rows that is written by one SQL statement. Default is 100000.
+ *
+ * @param batchSize maximum number of rows that is written by one SQL statement
+ */
+ public Write<T> withBatchSize(Integer batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return withBatchSize(ValueProvider.StaticValueProvider.of(batchSize));
+ }
+
+ /** Same as {@link #withBatchSize(Integer)} but accepting a ValueProvider. */
+ public Write<T> withBatchSize(ValueProvider<Integer> batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return toBuilder().setBatchSize(batchSize).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<T> input) {
+ DataSourceConfiguration dataSourceConfiguration =
+ Util.getRequiredArgument(
+ getDataSourceConfiguration(), "withDataSourceConfiguration() is required");
+ String table = Util.getRequiredArgument(getTable(), "withTable() is required");
+ UserDataMapper<T> userDataMapper =
+ Util.getRequiredArgument(getUserDataMapper(), "withUserDataMapper() is required");
+ int batchSize = Util.getArgumentWithDefault(getBatchSize(), DEFAULT_BATCH_SIZE);
+ checkArgument(batchSize > 0, "batchSize should be greater then 0");
+
+ input
+ .apply(
+ ParDo.of(
+ new DoFn<T, List<String>>() {
+ @ProcessElement
+ public void processElement(ProcessContext context) throws Exception {
+ context.output(userDataMapper.mapRow(context.element()));
+ }
+ }))
+ .setCoder(ListCoder.of(StringUtf8Coder.of()))
+ .apply(ParDo.of(new BatchFn<>(batchSize)))
+ .apply(ParDo.of(new WriteFn<Void>(dataSourceConfiguration, table)))
+ .setCoder(VoidCoder.of());
+
+ return PDone.in(input.getPipeline());
+ }
+
+ private static class BatchFn<ParameterT> extends DoFn<ParameterT, Iterable<ParameterT>> {
+ List<ParameterT> batch = new ArrayList<>();
+ int batchSize;
+
+ BatchFn(int batchSize) {
+ this.batchSize = batchSize;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ batch.add(context.element());
+ if (batch.size() >= batchSize) {
+ context.output(batch);
+ batch = new ArrayList<>();
+ }
+ }
+
+ @FinishBundle
+ public void finish(FinishBundleContext context) {
+ if (batch.size() > 0) {
+ context.output(batch, Instant.now(), GlobalWindow.INSTANCE);
+ batch = new ArrayList<>();
+ }
+ }
+ }
+
+ private static class WriteFn<OutputT> extends DoFn<Iterable<List<String>>, OutputT> {
+ DataSourceConfiguration dataSourceConfiguration;
+ String table;
+
+ WriteFn(DataSourceConfiguration dataSourceConfiguration, String table) {
+ this.dataSourceConfiguration = dataSourceConfiguration;
+ this.table = table;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) throws Exception {
+ DataSource dataSource = dataSourceConfiguration.getDataSource();
+
+ Connection conn = dataSource.getConnection();
+ try {
+ Statement stmt = conn.createStatement();
+ try (PipedOutputStream baseStream = new PipedOutputStream();
+ InputStream inputStream = new PipedInputStream(baseStream, BUFFER_SIZE)) {
+ ((com.singlestore.jdbc.Statement) ((DelegatingStatement) stmt).getInnermostDelegate())
+ .setNextLocalInfileInputStream(inputStream);
+
+ final Exception[] writeException = new Exception[1];
+
+ Thread dataWritingThread =
Review Comment:
This is somewhat unusual for me as well. @lukecwik do we have a pattern for something like this?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1297164176
Run Python PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1320488360
ok let's get this in so tests are continuously exercised
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1315710802
Run Java SingleStoreIO Performance Test
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314487523
Also, we have a dockerized jenkins tool to run the jenkins test locally: https://github.com/apache/beam/tree/master/.test-infra/dockerized-jenkins
I have not used it yet and it seems there are some issue currently. Once #24054 is fixed it should be resolved and you can also run performance tests locally using jenkins docker.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314502814
Run Java SingleStoreIO Performance Test
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1023362367
##########
.test-infra/jenkins/job_PerformanceTests_SingleStoreIO.groovy:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as common
+import Kubernetes
+import InfluxDBCredentialsHelper
+
+String jobName = "beam_PerformanceTests_SingleStoreIO"
+
+void waitForPodWithLabel(job, Kubernetes k8s, String label) {
+ job.steps {
+ shell("${k8s.KUBERNETES_DIR}/singlestore/wait-for-pod-with-label.sh ${label} 600")
+ }
+}
+
+void waitFor(job, Kubernetes k8s, String resource) {
+ job.steps {
+ shell("${k8s.KUBERNETES_DIR}/singlestore/wait-for.sh ${resource} 600")
+ }
+}
+
+job(jobName) {
+ common.setTopLevelMainJobProperties(delegate)
+ common.setAutoJob(delegate,'H H/6 * * *')
+ common.enablePhraseTriggeringFromPullRequest(
+ delegate,
+ 'Java SingleStoreIO Performance Test',
+ 'Run Java SingleStoreIO Performance Test')
+ InfluxDBCredentialsHelper.useCredentials(delegate)
+
+
+ String namespace = common.getKubernetesNamespace(jobName)
+ String kubeconfigPath = common.getKubeconfigLocationForNamespace(namespace)
+ Kubernetes k8s = Kubernetes.create(delegate, kubeconfigPath, namespace)
+
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-rbac.yaml"))
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-cluster-crd.yaml"))
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-operator.yaml"))
+ waitForPodWithLabel(delegate, k8s, "sdb-operator")
+
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-cluster.yaml"))
+ waitFor(delegate, k8s, "memsqlclusters.memsql.com")
+
+ String singlestoreHostName = "LOAD_BALANCER_IP"
+ k8s.loadBalancerIP("svc-sdb-cluster-ddl", singlestoreHostName)
+
+ Map pipelineOptions = [
+ tempRoot : 'gs://temp-storage-for-perf-tests',
+ project : 'apache-beam-testing',
+ runner : 'DataflowRunner',
+ singleStoreServerName : "\$${singlestoreHostName}",
+ singleStoreUsername : "admin",
+ singleStorePassword : "secretpass",
+ singleStorePort: "3306",
+ numberOfRecords: "100000",
Review Comment:
Wow, it is really pretty fast :)
Increased the number of rows to 10M.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1305489382
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1317228238
Run Java SingleStoreIO Performance Test
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1313950060
run seed job
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1024355506
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/SingleStoreIO.java:
##########
@@ -0,0 +1,955 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Filter;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle;
+import org.apache.beam.sdk.transforms.SerializableFunctions;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.util.Preconditions;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.commons.dbcp2.BasicDataSource;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * IO to read and write data on SingleStoreDB.
+ *
+ * <h3>Reading from SingleStoreDB datasource</h3>
+ *
+ * <p>SingleStoreIO source returns a bounded collection of {@code T} as a {@code PCollection<T>}. T
+ * is the type returned by the provided {@link RowMapper}.
+ *
+ * <p>To configure the SingleStoreDB source, you have to provide a {@link DataSourceConfiguration}
+ * using {@link DataSourceConfiguration#create(String)}(endpoint). Optionally, {@link
+ * DataSourceConfiguration#withUsername(String)} and {@link
+ * DataSourceConfiguration#withPassword(String)} allows you to define username and password.
+ *
+ * <p>For example:
+ *
+ * <pre>{@code
+ * pipeline.apply(SingleStoreIO.<KV<Integer, String>>read()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
+ * .withUsername("username")
+ * .withPassword("password"))
+ * .withQuery("select id, name from Person")
+ * .withRowMapper(new RowMapper<KV<Integer, String>>() {
+ * public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
+ * return KV.of(resultSet.getInt(1), resultSet.getString(2));
+ * }
+ * })
+ * );
+ * }</pre>
+ *
+ * <p>Query parameters can be configured using a user-provided {@link StatementPreparator}. For
+ * example:
+ *
+ * <pre>{@code
+ * pipeline.apply(SingleStoreIO.<KV<Integer, String>>read()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306"))
+ * .withQuery("select id,name from Person where name = ?")
+ * .withStatementPreparator(new StatementPreparator() {
+ * public void setParameters(PreparedStatement preparedStatement) throws Exception {
+ * preparedStatement.setString(1, "Darwin");
+ * }
+ * })
+ * .withRowMapper(new RowMapper<KV<Integer, String>>() {
+ * public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
+ * return KV.of(resultSet.getInt(1), resultSet.getString(2));
+ * }
+ * })
+ * );
+ * }</pre>
+ *
+ * <h4>Parallel reading from a SingleStoreDB datasource</h4>
+ *
+ * <p>SingleStoreIO supports partitioned reading of all data from a table. To enable this, use
+ * {@link SingleStoreIO#readWithPartitions()}. This way of data reading is preferred because of the
+ * performance reasons.
+ *
+ * <p>The following example shows usage of {@link SingleStoreIO#readWithPartitions()}
+ *
+ * <pre>{@code
+ * pipeline.apply(SingleStoreIO.<Row>readWithPartitions()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
+ * .withUsername("username")
+ * .withPassword("password"))
+ * .withTable("Person")
+ * .withRowMapper(new RowMapper<KV<Integer, String>>() {
+ * public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
+ * return KV.of(resultSet.getInt(1), resultSet.getString(2));
+ * }
+ * })
+ * );
+ * }</pre>
+ *
+ * <h3>Writing to SingleStoreDB datasource</h3>
+ *
+ * <p>SingleStoreIO supports writing records into a database. It writes a {@link PCollection} to the
+ * database by converting data to CSV and sending it to the database with LOAD DATA query.
+ *
+ * <p>Like the source, to configure the sink, you have to provide a {@link DataSourceConfiguration}.
+ *
+ * <pre>{@code
+ * pipeline
+ * .apply(...)
+ * .apply(SingleStoreIO.<KV<Integer, String>>write()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
+ * .withUsername("username")
+ * .withPassword("password"))
+ * .withStatement("insert into Person values(?, ?)")
+ * .withUserDataMapper(new UserDataMapper<KV<Integer, String>>() {
+ * @Override
+ * public List<String> mapRow(KV<Integer, String> element) {
+ * List<String> res = new ArrayList<>();
+ * res.add(element.getKey().toString());
+ * res.add(element.getValue());
+ * return res;
+ * }
+ * })
+ * );
+ * }</pre>
+ */
+final class SingleStoreIO {
Review Comment:
Yea, sorry
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314453629
Run Java SingleStoreIO Performance Test
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314534027
Sure :)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314515047
Run Java SingleStoreIO Performance Test
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1024226842
##########
.test-infra/jenkins/job_PerformanceTests_SingleStoreIO.groovy:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as common
+import Kubernetes
+import InfluxDBCredentialsHelper
+
+String jobName = "beam_PerformanceTests_SingleStoreIO"
+
+void waitForPodWithLabel(job, Kubernetes k8s, String label) {
+ job.steps {
+ shell("${k8s.KUBERNETES_DIR}/singlestore/wait-for-pod-with-label.sh ${label} 600")
+ }
+}
+
+void waitFor(job, Kubernetes k8s, String resource) {
+ job.steps {
+ shell("${k8s.KUBERNETES_DIR}/singlestore/wait-for.sh ${resource} 600")
+ }
+}
+
+job(jobName) {
+ common.setTopLevelMainJobProperties(delegate)
+ common.setAutoJob(delegate,'H H/6 * * *')
+ common.enablePhraseTriggeringFromPullRequest(
+ delegate,
+ 'Java SingleStoreIO Performance Test',
+ 'Run Java SingleStoreIO Performance Test')
+ InfluxDBCredentialsHelper.useCredentials(delegate)
+
+
+ String namespace = common.getKubernetesNamespace(jobName)
+ String kubeconfigPath = common.getKubeconfigLocationForNamespace(namespace)
+ Kubernetes k8s = Kubernetes.create(delegate, kubeconfigPath, namespace)
+
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-rbac.yaml"))
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-cluster-crd.yaml"))
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-operator.yaml"))
+ waitForPodWithLabel(delegate, k8s, "sdb-operator")
+
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-cluster.yaml"))
+ waitFor(delegate, k8s, "memsqlclusters.memsql.com")
+
+ String singlestoreHostName = "LOAD_BALANCER_IP"
+ k8s.loadBalancerIP("svc-sdb-cluster-ddl", singlestoreHostName)
+
+ Map pipelineOptions = [
+ tempRoot : 'gs://temp-storage-for-perf-tests',
+ project : 'apache-beam-testing',
+ runner : 'DataflowRunner',
+ singleStoreServerName : "\$${singlestoreHostName}",
+ singleStoreUsername : "admin",
+ singleStorePassword : "secretpass",
+ singleStorePort: "3306",
+ numberOfRecords: "100000",
Review Comment:
ah sorry. The test row pre-defined some hash for validation. 10000000 does not work but 5000000 will do:
```
private static final ImmutableMap<Integer, String> EXPECTED_HASHES =
ImmutableMap.of(
1000, "7d94d63a41164be058a9680002914358",
100_000, "c7cbddb319209e200f1c5eebef8fe960",
600_000, "e2add2f680de9024e9bc46cd3912545e",
5_000_000, "c44f8a5648cd9207c9c6f77395a998dc");
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1271452217
# [Codecov](https://codecov.io/gh/apache/beam/pull/23535?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#23535](https://codecov.io/gh/apache/beam/pull/23535?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (4952064) into [master](https://codecov.io/gh/apache/beam/commit/17fb9c0342064cd4375b0d7f2c37e12a175d03ef?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (17fb9c0) will **not change** coverage.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #23535 +/- ##
=======================================
Coverage 74.20% 74.20%
=======================================
Files 708 708
Lines 93462 93462
=======================================
Hits 69355 69355
Misses 22832 22832
Partials 1275 1275
```
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1294914593
R: @johnjcasey
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1294915441
R: @TheNeuralBit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1294912160
R: @kennknowles
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1023394717
##########
sdks/java/io/singlestore/src/test/java/org/apache/beam/sdk/io/singlestore/SingleStoreIOIT.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.sdk.io.common.IOITHelper.readIOTestPipelineOptions;
+
+import com.google.cloud.Timestamp;
+import com.singlestore.jdbc.SingleStoreDataSource;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.function.Function;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.io.common.DatabaseTestHelper;
+import org.apache.beam.sdk.io.common.HashingFn;
+import org.apache.beam.sdk.io.common.TestRow;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testutils.NamedTestResult;
+import org.apache.beam.sdk.testutils.metrics.IOITMetrics;
+import org.apache.beam.sdk.testutils.metrics.MetricsReader;
+import org.apache.beam.sdk.testutils.metrics.TimeMonitor;
+import org.apache.beam.sdk.testutils.publishing.InfluxDBSettings;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Sum;
+import org.apache.beam.sdk.transforms.Top;
+import org.apache.beam.sdk.values.PCollection;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class SingleStoreIOIT {
+
+ private static final String NAMESPACE = SingleStoreIOIT.class.getName();
+
+ private static final String DATABASE_NAME = "SingleStoreIOIT";
+
+ private static int numberOfRows;
+
+ private static String tableName;
+
+ private static String serverName;
+
+ private static String username;
+
+ private static String password;
+
+ private static Integer port;
+
+ private static SingleStoreIO.DataSourceConfiguration dataSourceConfiguration;
+
+ private static InfluxDBSettings settings;
+
+ @BeforeClass
+ public static void setup() {
+ SingleStoreIOTestPipelineOptions options;
+ try {
+ options = readIOTestPipelineOptions(SingleStoreIOTestPipelineOptions.class);
+ } catch (IllegalArgumentException e) {
+ options = null;
+ }
+ org.junit.Assume.assumeNotNull(options);
+
+ numberOfRows = options.getNumberOfRecords();
+ serverName = options.getSingleStoreServerName();
+ username = options.getSingleStoreUsername();
+ password = options.getSingleStorePassword();
+ port = options.getSingleStorePort();
+ tableName = DatabaseTestHelper.getTestTableName("IT");
+ dataSourceConfiguration =
+ SingleStoreIO.DataSourceConfiguration.create(serverName + ":" + port)
+ .withDatabase(DATABASE_NAME)
+ .withPassword(password)
+ .withUsername(username);
+ settings =
+ InfluxDBSettings.builder()
+ .withHost(options.getInfluxHost())
+ .withDatabase(options.getInfluxDatabase())
+ .withMeasurement(options.getInfluxMeasurement())
+ .get();
+ }
+
+ void createDatabaseIfNotExists() throws SQLException {
+ DataSource dataSource =
+ new SingleStoreDataSource(
+ String.format(
+ "jdbc:singlestore://%s:%d/?user=%s&password=%s&allowLocalInfile=TRUE",
+ serverName, port, username, password));
+ try (Connection conn = dataSource.getConnection();
+ Statement stmt = conn.createStatement()) {
+ stmt.executeQuery(String.format("CREATE DATABASE IF NOT EXISTS %s", DATABASE_NAME));
+ }
+ }
+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testWriteThenRead() throws Exception {
+ createDatabaseIfNotExists();
+ DataSource dataSource =
+ new SingleStoreDataSource(
+ String.format(
+ "jdbc:singlestore://%s:%d/%s?user=%s&password=%s&allowLocalInfile=TRUE",
+ serverName, port, DATABASE_NAME, username, password));
+ DatabaseTestHelper.createTable(dataSource, tableName);
+ try {
+ PipelineResult writeResult = runWrite();
+ writeResult.waitUntilFinish();
+ PipelineResult readResult = runRead();
+ readResult.waitUntilFinish();
Review Comment:
number of records is checked here
```
private void testReadResult(PCollection<TestRow> namesAndIds) {
PAssert.thatSingleton(namesAndIds.apply("Count All", Count.globally()))
.isEqualTo((long) numberOfRows);
```
(lines 248-250)
Will add asserts for pipeline status. Thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314508671
I thought that `SingleStore_IO_Direct` will run unit tests.
But after your comment, I realized that it is running the same integration test.
Deleted it for now.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1312173844
Thanks! Will take a look. For now, could you please integrate Read.java and Write.java into SingleStoreIO.java, as the patterns used for other IO connectors (e.g. [JdbcIO](https://github.com/apache/beam/blob/master/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcIO.java)). It users may not able to do something like
```
import org.apache.beam.sdk.io.singlestore.Read;
import org.apache.beam.sdk.io.singlestore.Write;
```
since there may be naming conflictions.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1297181987
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1018009823
##########
sdks/java/io/singlestore/build.gradle:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(
+ automaticModuleName: 'org.apache.beam.sdk.io.singlestore',
+)
+provideIntegrationTestingDependencies()
+enableJavaPerformanceTesting()
+
+description = "Apache Beam :: SDKs :: Java :: IO :: SingleStore"
+ext.summary = "IO to read and write on SingleStore datasource."
+
+dependencies {
+ implementation library.java.vendored_guava_26_0_jre
+ implementation library.java.joda_time
+ implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation group: 'com.singlestore', name: 'singlestore-jdbc-client', version: '1.1.4'
+ implementation library.java.slf4j_api
+ implementation "org.apache.commons:commons-dbcp2:2.8.0"
Review Comment:
Do I understand correctly that your proposition is to add these libs to this map
```
// A map of maps containing common libraries used per language. To use:
// dependencies {
// compile library.java.slf4j_api
// }
project.ext.library = [
java : [
```
and use it 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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] johnjcasey commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
johnjcasey commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1018067643
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/Write.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VoidCoder;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+
+@AutoValue
+public abstract class Write<T> extends PTransform<PCollection<T>, PDone> {
+
+ private static final int DEFAULT_BATCH_SIZE = 100000;
+ private static final int BUFFER_SIZE = 524288;
+
+ abstract @Nullable DataSourceConfiguration getDataSourceConfiguration();
+
+ abstract @Nullable ValueProvider<String> getTable();
+
+ abstract @Nullable ValueProvider<Integer> getBatchSize();
+
+ abstract @Nullable UserDataMapper<T> getUserDataMapper();
+
+ abstract Builder<T> toBuilder();
+
+ @AutoValue.Builder
+ abstract static class Builder<T> {
+ abstract Builder<T> setDataSourceConfiguration(DataSourceConfiguration dataSourceConfiguration);
+
+ abstract Builder<T> setTable(ValueProvider<String> table);
+
+ abstract Builder<T> setBatchSize(ValueProvider<Integer> batchSize);
+
+ abstract Builder<T> setUserDataMapper(UserDataMapper<T> userDataMapper);
+
+ abstract Write<T> build();
+ }
+
+ public Write<T> withDataSourceConfiguration(DataSourceConfiguration config) {
+ checkNotNull(config, "dataSourceConfiguration can not be null");
+ return toBuilder().setDataSourceConfiguration(config).build();
+ }
+
+ public Write<T> withTable(String table) {
+ checkNotNull(table, "table can not be null");
+ return withTable(ValueProvider.StaticValueProvider.of(table));
+ }
+
+ public Write<T> withTable(ValueProvider<String> table) {
+ checkNotNull(table, "table can not be null");
+ return toBuilder().setTable(table).build();
+ }
+
+ public Write<T> withUserDataMapper(UserDataMapper<T> userDataMapper) {
+ checkNotNull(userDataMapper, "userDataMapper can not be null");
+ return toBuilder().setUserDataMapper(userDataMapper).build();
+ }
+
+ /**
+ * Provide a maximum number of rows that is written by one SQL statement. Default is 100000.
+ *
+ * @param batchSize maximum number of rows that is written by one SQL statement
+ */
+ public Write<T> withBatchSize(Integer batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return withBatchSize(ValueProvider.StaticValueProvider.of(batchSize));
+ }
+
+ /** Same as {@link #withBatchSize(Integer)} but accepting a ValueProvider. */
+ public Write<T> withBatchSize(ValueProvider<Integer> batchSize) {
+ checkNotNull(batchSize, "batchSize can not be null");
+ return toBuilder().setBatchSize(batchSize).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<T> input) {
+ DataSourceConfiguration dataSourceConfiguration =
+ Util.getRequiredArgument(
+ getDataSourceConfiguration(), "withDataSourceConfiguration() is required");
+ String table = Util.getRequiredArgument(getTable(), "withTable() is required");
+ UserDataMapper<T> userDataMapper =
+ Util.getRequiredArgument(getUserDataMapper(), "withUserDataMapper() is required");
+ int batchSize = Util.getArgumentWithDefault(getBatchSize(), DEFAULT_BATCH_SIZE);
+ checkArgument(batchSize > 0, "batchSize should be greater then 0");
+
+ input
+ .apply(
+ ParDo.of(
+ new DoFn<T, List<String>>() {
+ @ProcessElement
+ public void processElement(ProcessContext context) throws Exception {
+ context.output(userDataMapper.mapRow(context.element()));
+ }
+ }))
+ .setCoder(ListCoder.of(StringUtf8Coder.of()))
+ .apply(ParDo.of(new BatchFn<>(batchSize)))
+ .apply(ParDo.of(new WriteFn<Void>(dataSourceConfiguration, table)))
Review Comment:
sounds good. Its also frequently useful to return any errors that may have been thrown by the query, especially if those aren't transient
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AdalbertMemSQL commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
AdalbertMemSQL commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314382578
Hi @Abacn
Looks like I was able to rebase it to the latest master but I'm not sure that I did it in the correct way.
Sorry, if I did something wrong. This is one of my first contributions to the non-SingleStore open-source projects.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1023138062
##########
CHANGES.md:
##########
@@ -60,6 +60,7 @@
* Support for Bigtable sink (Write and WriteBatch) added (Go) ([#23324](https://github.com/apache/beam/issues/23324)).
* S3 implementation of the Beam filesystem (Go) ([#23991](https://github.com/apache/beam/issues/23991)).
+* Support for SingleStoreDB source added (Java) ([#22617](https://github.com/apache/beam/issues/22617)).
Review Comment:
Do you mean "SingleStoreDB Source and Sink"?
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/SingleStoreUtil.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import org.apache.beam.sdk.coders.CannotProvideCoderException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderRegistry;
+import org.apache.beam.sdk.schemas.NoSuchSchemaException;
+import org.apache.beam.sdk.schemas.SchemaRegistry;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+
+/** Provides utility functions for working with {@link SingleStoreIO}. */
+public class SingleStoreUtil {
Review Comment:
```suggestion
final class SingleStoreUtil {
```
We may not expose this as public API (as the practice of other IOs)
##########
.test-infra/jenkins/job_PerformanceTests_SingleStoreIO.groovy:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as common
+import Kubernetes
+import InfluxDBCredentialsHelper
+
+String jobName = "beam_PerformanceTests_SingleStoreIO"
+
+void waitForPodWithLabel(job, Kubernetes k8s, String label) {
+ job.steps {
+ shell("${k8s.KUBERNETES_DIR}/singlestore/wait-for-pod-with-label.sh ${label} 600")
+ }
+}
+
+void waitFor(job, Kubernetes k8s, String resource) {
+ job.steps {
+ shell("${k8s.KUBERNETES_DIR}/singlestore/wait-for.sh ${resource} 600")
+ }
+}
+
+job(jobName) {
+ common.setTopLevelMainJobProperties(delegate)
+ common.setAutoJob(delegate,'H H/6 * * *')
+ common.enablePhraseTriggeringFromPullRequest(
+ delegate,
+ 'Java SingleStoreIO Performance Test',
+ 'Run Java SingleStoreIO Performance Test')
+ InfluxDBCredentialsHelper.useCredentials(delegate)
+
+
+ String namespace = common.getKubernetesNamespace(jobName)
+ String kubeconfigPath = common.getKubeconfigLocationForNamespace(namespace)
+ Kubernetes k8s = Kubernetes.create(delegate, kubeconfigPath, namespace)
+
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-rbac.yaml"))
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-cluster-crd.yaml"))
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-operator.yaml"))
+ waitForPodWithLabel(delegate, k8s, "sdb-operator")
+
+ k8s.apply(common.makePathAbsolute("src/.test-infra/kubernetes/singlestore/sdb-cluster.yaml"))
+ waitFor(delegate, k8s, "memsqlclusters.memsql.com")
+
+ String singlestoreHostName = "LOAD_BALANCER_IP"
+ k8s.loadBalancerIP("svc-sdb-cluster-ddl", singlestoreHostName)
+
+ Map pipelineOptions = [
+ tempRoot : 'gs://temp-storage-for-perf-tests',
+ project : 'apache-beam-testing',
+ runner : 'DataflowRunner',
+ singleStoreServerName : "\$${singlestoreHostName}",
+ singleStoreUsername : "admin",
+ singleStorePassword : "secretpass",
+ singleStorePort: "3306",
+ numberOfRecords: "100000",
Review Comment:
Looks like the performance test is passing. The Write then Read steps finishing within seconds. To get a more meaningful metric, ideally the pipeline runs several minutes. How about setting numberOfRecords to 10000000?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1315599929
Run Seed Job
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314451576
Hi, no worry. Basically If changes involve .test-infra folder I need to run a seed job on my side to make the changes take effect.
Get precommit test having error
```
16:43:16 ERROR: Step ‘Publish JUnit test result report’ failed: No test report files were found. Configuration error?
```
https://ci-beam.apache.org/job/beam_PreCommit_Java_SingleStore_IO_Direct_Commit/1/console
performance tests did not finish with 'Pod is Running'
Feel free to pin me if you want to run the seedjob again.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] johnjcasey commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
johnjcasey commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1312166181
This looks good to me outside of the threading, which I'm unsure about.
@chamikaramj and @Abacn can you take a second look, as this is an entire IO I'd like more pairs of eyes
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1294913351
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on a diff in pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on code in PR #23535:
URL: https://github.com/apache/beam/pull/23535#discussion_r1024352346
##########
sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/SingleStoreIO.java:
##########
@@ -0,0 +1,955 @@
+/*
+ * 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.beam.sdk.io.singlestore;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.ListCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Filter;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle;
+import org.apache.beam.sdk.transforms.SerializableFunctions;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.util.Preconditions;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.commons.dbcp2.BasicDataSource;
+import org.apache.commons.dbcp2.DelegatingStatement;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * IO to read and write data on SingleStoreDB.
+ *
+ * <h3>Reading from SingleStoreDB datasource</h3>
+ *
+ * <p>SingleStoreIO source returns a bounded collection of {@code T} as a {@code PCollection<T>}. T
+ * is the type returned by the provided {@link RowMapper}.
+ *
+ * <p>To configure the SingleStoreDB source, you have to provide a {@link DataSourceConfiguration}
+ * using {@link DataSourceConfiguration#create(String)}(endpoint). Optionally, {@link
+ * DataSourceConfiguration#withUsername(String)} and {@link
+ * DataSourceConfiguration#withPassword(String)} allows you to define username and password.
+ *
+ * <p>For example:
+ *
+ * <pre>{@code
+ * pipeline.apply(SingleStoreIO.<KV<Integer, String>>read()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
+ * .withUsername("username")
+ * .withPassword("password"))
+ * .withQuery("select id, name from Person")
+ * .withRowMapper(new RowMapper<KV<Integer, String>>() {
+ * public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
+ * return KV.of(resultSet.getInt(1), resultSet.getString(2));
+ * }
+ * })
+ * );
+ * }</pre>
+ *
+ * <p>Query parameters can be configured using a user-provided {@link StatementPreparator}. For
+ * example:
+ *
+ * <pre>{@code
+ * pipeline.apply(SingleStoreIO.<KV<Integer, String>>read()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306"))
+ * .withQuery("select id,name from Person where name = ?")
+ * .withStatementPreparator(new StatementPreparator() {
+ * public void setParameters(PreparedStatement preparedStatement) throws Exception {
+ * preparedStatement.setString(1, "Darwin");
+ * }
+ * })
+ * .withRowMapper(new RowMapper<KV<Integer, String>>() {
+ * public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
+ * return KV.of(resultSet.getInt(1), resultSet.getString(2));
+ * }
+ * })
+ * );
+ * }</pre>
+ *
+ * <h4>Parallel reading from a SingleStoreDB datasource</h4>
+ *
+ * <p>SingleStoreIO supports partitioned reading of all data from a table. To enable this, use
+ * {@link SingleStoreIO#readWithPartitions()}. This way of data reading is preferred because of the
+ * performance reasons.
+ *
+ * <p>The following example shows usage of {@link SingleStoreIO#readWithPartitions()}
+ *
+ * <pre>{@code
+ * pipeline.apply(SingleStoreIO.<Row>readWithPartitions()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
+ * .withUsername("username")
+ * .withPassword("password"))
+ * .withTable("Person")
+ * .withRowMapper(new RowMapper<KV<Integer, String>>() {
+ * public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
+ * return KV.of(resultSet.getInt(1), resultSet.getString(2));
+ * }
+ * })
+ * );
+ * }</pre>
+ *
+ * <h3>Writing to SingleStoreDB datasource</h3>
+ *
+ * <p>SingleStoreIO supports writing records into a database. It writes a {@link PCollection} to the
+ * database by converting data to CSV and sending it to the database with LOAD DATA query.
+ *
+ * <p>Like the source, to configure the sink, you have to provide a {@link DataSourceConfiguration}.
+ *
+ * <pre>{@code
+ * pipeline
+ * .apply(...)
+ * .apply(SingleStoreIO.<KV<Integer, String>>write()
+ * .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
+ * .withUsername("username")
+ * .withPassword("password"))
+ * .withStatement("insert into Person values(?, ?)")
+ * .withUserDataMapper(new UserDataMapper<KV<Integer, String>>() {
+ * @Override
+ * public List<String> mapRow(KV<Integer, String> element) {
+ * List<String> res = new ArrayList<>();
+ * res.add(element.getKey().toString());
+ * res.add(element.getValue());
+ * return res;
+ * }
+ * })
+ * );
+ * }</pre>
+ */
+final class SingleStoreIO {
Review Comment:
Oops I believe this is a mistake. `final class SingleStoreUtil` and `public class SingleStoreUtil`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1317357953
Run Java SingleStoreIO Performance Test
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1317350270
Run Java SingleStoreIO Performance Test
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1317556496
Let us pin @lukecwik and @chamikaramj again who got mentioned in this thread and see if they have any inputs. Will be a nice feature for Beam 2.44.0
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1317229278
Run Java_Examples_Dataflow_Java17 PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314033369
Run SQL PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #23535: SingleStoreIO
Posted by GitBox <gi...@apache.org>.
Abacn commented on PR #23535:
URL: https://github.com/apache/beam/pull/23535#issuecomment-1314505583
Run Java_Spark3_Versions PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org