This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 23da742275 [core] Enforce LIMIT at the table-read level for
primary-key tables (#8116)
23da742275 is described below
commit 23da742275ba10d5e4dabecaba68bb5a45fc1113
Author: wangwj <[email protected]>
AuthorDate: Mon Jul 13 12:06:20 2026 +0800
[core] Enforce LIMIT at the table-read level for primary-key tables (#8116)
---
.../apache/paimon/reader/LimitRecordReader.java | 92 +++++++++
.../paimon/table/format/FormatTableRead.java | 48 +----
.../paimon/table/source/KeyValueTableRead.java | 15 +-
.../paimon/table/PrimaryKeySimpleTableTest.java | 229 +++++++++++++++++++++
.../paimon/flink/KeyValueTableReadLimitITCase.java | 110 ++++++++++
5 files changed, 444 insertions(+), 50 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java
b/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java
new file mode 100644
index 0000000000..59e1a767b8
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.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.paimon.reader;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** A {@link RecordReader} that stops after reading a given number of records.
*/
+public final class LimitRecordReader<T> implements RecordReader<T> {
+
+ private final RecordReader<T> reader;
+ private final long limit;
+ private final AtomicLong recordCount = new AtomicLong(0);
+
+ public LimitRecordReader(RecordReader<T> reader, long limit) {
+ checkArgument(limit > 0, "Limit must be positive.");
+ this.reader = reader;
+ this.limit = limit;
+ }
+
+ public static <T> RecordReader<T> limit(RecordReader<T> reader, @Nullable
Integer limit) {
+ if (limit == null || limit <= 0) {
+ return reader;
+ }
+ return new LimitRecordReader<>(reader, limit);
+ }
+
+ @Override
+ @Nullable
+ public RecordIterator<T> readBatch() throws IOException {
+ if (recordCount.get() >= limit) {
+ return null;
+ }
+ RecordIterator<T> iterator = reader.readBatch();
+ if (iterator == null) {
+ return null;
+ }
+ return new LimitRecordIterator<>(iterator);
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+
+ private class LimitRecordIterator<T> implements RecordIterator<T> {
+
+ private final RecordIterator<T> iterator;
+
+ private LimitRecordIterator(RecordIterator<T> iterator) {
+ this.iterator = iterator;
+ }
+
+ @Override
+ @Nullable
+ public T next() throws IOException {
+ if (recordCount.get() >= limit) {
+ return null;
+ }
+ T next = iterator.next();
+ if (next != null) {
+ recordCount.incrementAndGet();
+ }
+ return next;
+ }
+
+ @Override
+ public void releaseBatch() {
+ iterator.releaseBatch();
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
index 765cfd2c9a..9939b1b750 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
@@ -23,6 +23,7 @@ import org.apache.paimon.disk.IOManager;
import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateProjectionConverter;
+import org.apache.paimon.reader.LimitRecordReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.source.Split;
@@ -31,7 +32,6 @@ import org.apache.paimon.types.RowType;
import java.io.IOException;
import java.util.Optional;
-import java.util.concurrent.atomic.AtomicLong;
/** A {@link TableRead} implementation for {@link FormatTable}. */
public class FormatTableRead implements TableRead {
@@ -80,11 +80,7 @@ public class FormatTableRead implements TableRead {
if (executeFilter) {
reader = executeFilter(reader);
}
- if (limit != null && limit > 0) {
- reader = applyLimit(reader, limit);
- }
-
- return reader;
+ return LimitRecordReader.limit(reader, limit);
}
private RecordReader<InternalRow> executeFilter(RecordReader<InternalRow>
reader) {
@@ -106,44 +102,4 @@ public class FormatTableRead implements TableRead {
Predicate finalFilter = predicate;
return reader.filter(finalFilter::test);
}
-
- private RecordReader<InternalRow> applyLimit(RecordReader<InternalRow>
reader, int limit) {
- return new RecordReader<InternalRow>() {
- private final AtomicLong recordCount = new AtomicLong(0);
-
- @Override
- public RecordIterator<InternalRow> readBatch() throws IOException {
- if (recordCount.get() >= limit) {
- return null;
- }
- RecordIterator<InternalRow> iterator = reader.readBatch();
- if (iterator == null) {
- return null;
- }
- return new RecordIterator<InternalRow>() {
- @Override
- public InternalRow next() throws IOException {
- if (recordCount.get() >= limit) {
- return null;
- }
- InternalRow next = iterator.next();
- if (next != null) {
- recordCount.incrementAndGet();
- }
- return next;
- }
-
- @Override
- public void releaseBatch() {
- iterator.releaseBatch();
- }
- };
- }
-
- @Override
- public void close() throws IOException {
- reader.close();
- }
- };
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
index 48b08df0c1..0b19df5c0c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
@@ -28,6 +28,7 @@ import org.apache.paimon.operation.RawFileSplitRead;
import org.apache.paimon.operation.SplitRead;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.TopN;
+import org.apache.paimon.reader.LimitRecordReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.TableSchema;
import
org.apache.paimon.table.source.splitread.IncrementalChangelogReadProvider;
@@ -96,9 +97,6 @@ public final class KeyValueTableRead extends
AbstractDataTableRead {
if (topN != null) {
read = read.withTopN(topN);
}
- if (limit != null) {
- read = read.withLimit(limit);
- }
read.withFilter(predicate).withIOManager(ioManager);
}
@@ -131,11 +129,20 @@ public final class KeyValueTableRead extends
AbstractDataTableRead {
@Override
public InnerTableRead withLimit(int limit) {
- initialized().forEach(r -> r.withLimit(limit));
this.limit = limit;
return this;
}
+ @Override
+ public RecordReader<InternalRow> createReader(List<Split> splits) throws
IOException {
+ return LimitRecordReader.limit(super.createReader(splits), limit);
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(Split split) throws
IOException {
+ return LimitRecordReader.limit(super.createReader(split), limit);
+ }
+
@Override
public TableRead withIOManager(IOManager ioManager) {
initialized().forEach(r -> r.withIOManager(ioManager));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index 245eb6ccf0..dd3a21b86c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.ChangelogProducer;
import org.apache.paimon.KeyValue;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.TableQueryAuthResult;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
@@ -64,6 +65,7 @@ import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.table.sink.WriteSelector;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.QueryAuthSplit;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.table.source.Split;
@@ -80,6 +82,7 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.ChangelogManager;
+import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.RoaringBitmap32;
@@ -3365,4 +3368,230 @@ public class PrimaryKeySimpleTableTest extends
SimpleTableTestBase {
write.close();
commit.close();
}
+
+ @Test
+ public void testReadWithLimitOnEmptyPlan() throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ TableScan.Plan plan = table.newScan().plan();
+ assertThat(plan.splits()).isEmpty();
+
+ RecordReader<InternalRow> reader =
+ table.newRead().withLimit(10).createReader(plan.splits());
+ AtomicInteger cnt = new AtomicInteger(0);
+ reader.forEachRemaining(row -> cnt.incrementAndGet());
+ assertThat(cnt.get()).isEqualTo(0);
+ reader.close();
+ }
+
+ @Test
+ public void testReadWithLimitThroughTableReadPath() throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ // Write multiple times to create overlapping level-0 files in the
same bucket.
+ int numRecords = 100;
+ for (int i = 0; i < numRecords; i++) {
+ write.write(rowData(1, i, (long) i));
+ }
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ for (int i = 0; i < numRecords; i++) {
+ write.write(rowData(1, i, (long) (i + 1)));
+ }
+ commit.commit(1, write.prepareCommit(true, 1));
+ write.close();
+ commit.close();
+
+ // Verify that withLimit takes effect through the table read path.
+ int limit = 10;
+ TableScan.Plan plan = table.newScan().plan();
+ RecordReader<InternalRow> reader =
+ table.newRead().withLimit(limit).createReader(plan.splits());
+ AtomicInteger cnt = new AtomicInteger(0);
+ reader.forEachRemaining(row -> cnt.incrementAndGet());
+ assertThat(cnt.get()).isEqualTo(limit);
+ reader.close();
+ }
+
+ @Test
+ public void testReadWithLimitThroughTableReadPathMultiSplit() throws
Exception {
+ FileStoreTable table = createFileStoreTable(conf -> conf.set(BUCKET,
4));
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ int numRecords = 50;
+ for (int i = 0; i < numRecords; i++) {
+ write.write(rowData(1, i, (long) i));
+ }
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ for (int i = 0; i < numRecords; i++) {
+ write.write(rowData(1, i, (long) (i + 1)));
+ }
+ commit.commit(1, write.prepareCommit(true, 1));
+ write.close();
+ commit.close();
+
+ TableScan.Plan plan = table.newScan().plan();
+ assertThat(plan.splits().size()).isGreaterThan(1);
+
+ int limit = 10;
+ RecordReader<InternalRow> reader =
+ table.newRead().withLimit(limit).createReader(plan.splits());
+ AtomicInteger cnt = new AtomicInteger(0);
+ reader.forEachRemaining(row -> cnt.incrementAndGet());
+ assertThat(cnt.get()).isEqualTo(limit);
+ reader.close();
+ }
+
+ @Test
+ public void testReadWithLimitThroughTableReadPathWithNonPrimaryKeyFilter()
throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ int numRecords = 100;
+ for (int i = 0; i < numRecords; i++) {
+ write.write(rowData(1, i, (long) i));
+ }
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ for (int i = 0; i < numRecords; i++) {
+ write.write(rowData(1, i, (long) (i + 1)));
+ }
+ commit.commit(1, write.prepareCommit(true, 1));
+ write.close();
+ commit.close();
+
+ int limit = 10;
+ Predicate nonPrimaryKeyFilter = new
PredicateBuilder(ROW_TYPE).isNotNull(2);
+ TableScan.Plan plan =
table.newScan().withFilter(nonPrimaryKeyFilter).plan();
+ assertThat(plan.splits()).hasSize(1);
+
+ RecordReader<InternalRow> reader =
+ table.newRead()
+ .withFilter(nonPrimaryKeyFilter)
+ .withLimit(limit)
+ .createReader(plan.splits());
+ AtomicInteger cnt = new AtomicInteger(0);
+ reader.forEachRemaining(row -> cnt.incrementAndGet());
+ assertThat(cnt.get()).isEqualTo(limit);
+ reader.close();
+ }
+
+ @Test
+ public void testReadWithLimitThroughTableReadPathWithQueryAuthFilter()
throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ writeOverlappingL0Records(table, 100);
+
+ int limit = 10;
+ Predicate authFilter = new
PredicateBuilder(ROW_TYPE).greaterOrEqual(1, 90);
+ TableQueryAuthResult authResult = authResultWithFilter(authFilter);
+
+ TableScan.Plan plan = table.newScan().plan();
+ assertThat(plan.splits()).hasSize(1);
+ List<Split> authSplits = wrapWithAuth(plan.splits(), authResult);
+
+ RecordReader<InternalRow> reader =
+ table.newRead().withLimit(limit).createReader(authSplits);
+ AtomicInteger cnt = new AtomicInteger(0);
+ reader.forEachRemaining(row -> cnt.incrementAndGet());
+ assertThat(cnt.get()).isEqualTo(limit);
+ reader.close();
+ }
+
+ @Test
+ public void
testReadWithLimitThroughSingleSplitCreateReaderWithQueryAuthFilter()
+ throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ writeOverlappingL0Records(table, 100);
+
+ int limit = 10;
+ Predicate authFilter = new
PredicateBuilder(ROW_TYPE).greaterOrEqual(1, 50);
+ TableQueryAuthResult authResult = authResultWithFilter(authFilter);
+
+ TableScan.Plan plan = table.newScan().plan();
+ assertThat(plan.splits()).hasSize(1);
+ Split authSplit = wrapWithAuth(plan.splits(), authResult).get(0);
+
+ RecordReader<InternalRow> reader =
table.newRead().withLimit(limit).createReader(authSplit);
+ AtomicInteger cnt = new AtomicInteger(0);
+ reader.forEachRemaining(row -> cnt.incrementAndGet());
+ assertThat(cnt.get()).isEqualTo(limit);
+ reader.close();
+ }
+
+ @Test
+ public void
testReadWithLimitThroughSingleSplitCreateReaderWithNonPrimaryKeyFilter()
+ throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ writeOverlappingL0Records(table, 100);
+
+ int limit = 10;
+ Predicate nonPrimaryKeyFilter = new
PredicateBuilder(ROW_TYPE).isNotNull(2);
+ TableScan.Plan plan =
table.newScan().withFilter(nonPrimaryKeyFilter).plan();
+ assertThat(plan.splits()).hasSize(1);
+ Split split = plan.splits().get(0);
+
+ RecordReader<InternalRow> reader =
+ table.newRead()
+ .withFilter(nonPrimaryKeyFilter)
+ .withLimit(limit)
+ .createReader(split);
+ AtomicInteger cnt = new AtomicInteger(0);
+ reader.forEachRemaining(row -> cnt.incrementAndGet());
+ assertThat(cnt.get()).isEqualTo(limit);
+ reader.close();
+ }
+
+ @Test
+ public void
testReadWithLimitThroughTableReadPathMultiSplitWithQueryAuthFilter()
+ throws Exception {
+ FileStoreTable table = createFileStoreTable(conf -> conf.set(BUCKET,
4));
+ writeOverlappingL0Records(table, 50);
+
+ int limit = 10;
+ Predicate authFilter = new
PredicateBuilder(ROW_TYPE).greaterOrEqual(1, 40);
+ TableQueryAuthResult authResult = authResultWithFilter(authFilter);
+
+ TableScan.Plan plan = table.newScan().plan();
+ assertThat(plan.splits().size()).isGreaterThan(1);
+ List<Split> authSplits = wrapWithAuth(plan.splits(), authResult);
+
+ RecordReader<InternalRow> reader =
+ table.newRead().withLimit(limit).createReader(authSplits);
+ AtomicInteger cnt = new AtomicInteger(0);
+ reader.forEachRemaining(row -> cnt.incrementAndGet());
+ assertThat(cnt.get()).isEqualTo(limit);
+ reader.close();
+ }
+
+ private void writeOverlappingL0Records(FileStoreTable table, int
numRecords) throws Exception {
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ for (int i = 0; i < numRecords; i++) {
+ write.write(rowData(1, i, (long) i));
+ }
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ for (int i = 0; i < numRecords; i++) {
+ write.write(rowData(1, i, (long) (i + 1)));
+ }
+ commit.commit(1, write.prepareCommit(true, 1));
+ write.close();
+ commit.close();
+ }
+
+ private TableQueryAuthResult authResultWithFilter(Predicate predicate) {
+ return new TableQueryAuthResult(
+
Collections.singletonList(JsonSerdeUtil.toFlatJson(predicate)), null);
+ }
+
+ private List<Split> wrapWithAuth(List<Split> splits, TableQueryAuthResult
authResult) {
+ return splits.stream()
+ .map(split -> new QueryAuthSplit(split, authResult))
+ .collect(Collectors.toList());
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/KeyValueTableReadLimitITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/KeyValueTableReadLimitITCase.java
new file mode 100644
index 0000000000..6428394c67
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/KeyValueTableReadLimitITCase.java
@@ -0,0 +1,110 @@
+/*
+ * 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.paimon.flink;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.stream.IntStream;
+
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Batch IT for TableRead LIMIT on primary key tables with multiple level-0
files. */
+public class KeyValueTableReadLimitITCase extends CatalogITCaseBase {
+
+ private static final int TOTAL_ROWS = 100;
+ private static final int COMMITS = 5;
+ private static final int LIMIT = 10;
+
+ @Override
+ protected List<String> ddl() {
+ return singletonList(
+ "CREATE TABLE IF NOT EXISTS pk_l0_limit ("
+ + "id INT, v INT, PRIMARY KEY (id) NOT ENFORCED) "
+ + "WITH ("
+ + "'bucket' = '1', "
+ + "'write-only' = 'true', "
+ + "'changelog-producer' = 'none', "
+ + "'merge-engine' = 'deduplicate')");
+ }
+
+ @Nullable
+ @Override
+ protected Boolean sqlSyncMode() {
+ return true;
+ }
+
+ @Test
+ public void testLimitOnMultiLevel0PrimaryKeyTable() {
+ writeMultipleCommits();
+
+ long level0FileCount =
+ batchSql("SELECT file_path FROM `pk_l0_limit$files` WHERE
level = 0").size();
+ assertThat(level0FileCount)
+ .as("Need multiple level-0 files to exercise merge read path")
+ .isGreaterThan(1);
+
+ List<Row> limited =
+ batchSql(
+ String.format(
+ "SELECT * FROM pk_l0_limit /*+
OPTIONS('scan.infer-parallelism'='false', 'scan.parallelism'='1') */ LIMIT %d",
+ LIMIT));
+ assertThat(limited).hasSize(LIMIT);
+
+ List<Row> all = batchSql("SELECT * FROM pk_l0_limit");
+ assertThat(all).hasSize(TOTAL_ROWS);
+ }
+
+ @Test
+ public void testLimitWithValueFilterReturnsCorrectRows() {
+ writeMultipleCommits();
+
+ List<Row> filtered =
+ batchSql(
+ String.format(
+ "SELECT * FROM pk_l0_limit /*+
OPTIONS('scan.infer-parallelism'='false', 'scan.parallelism'='1') */ "
+ + "WHERE v > %d LIMIT %d",
+ TOTAL_ROWS / 2, LIMIT));
+ assertThat(filtered).hasSize(LIMIT);
+ for (Row row : filtered) {
+ assertThat((int) row.getField(1)).isGreaterThan(TOTAL_ROWS / 2);
+ }
+ }
+
+ private void writeMultipleCommits() {
+ int rowsPerCommit = TOTAL_ROWS / COMMITS;
+ IntStream.range(0, COMMITS)
+ .forEach(
+ commit -> {
+ StringBuilder values = new StringBuilder();
+ for (int i = 0; i < rowsPerCommit; i++) {
+ int id = commit * rowsPerCommit + i;
+ if (i > 0) {
+ values.append(", ");
+ }
+ values.append(String.format("(%d, %d)", id,
id));
+ }
+ batchSql("INSERT INTO pk_l0_limit VALUES " +
values);
+ });
+ }
+}
