Skip to content

Bump lance version to 5.0.0-beta.6 and arrow to 15.0.2 #20

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SteveStevenpoor wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Bump lance version to 5.0.0-beta.6 and arrow to 15.0.2 #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,9 @@

<!-- 版本管理 -->
<flink.version>1.16.1</flink.version>
<lance.version>0.23.3</lance.version>
<arrow.version>14.0.0</arrow.version>
<junit.version>5.9.3</junit.version>
<lance.version>5.0.0-beta.6</lance.version>
<arrow.version>15.0.2</arrow.version>
<junit.version>5.10.1</junit.version>
<slf4j.version>1.7.36</slf4j.version>
<log4j.version>2.20.0</log4j.version>
<mockito.version>5.3.1</mockito.version>
Expand Down Expand Up @@ -73,9 +73,15 @@

<!-- ==================== Lance Java SDK ==================== -->
<dependency>
<groupId>com.lancedb</groupId>
<groupId>org.lance</groupId>
<artifactId>lance-core</artifactId>
<version>${lance.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
</exclusions>
</dependency>

<!-- ==================== Apache Arrow ==================== -->
Expand Down
10 changes: 5 additions & 5 deletions src/main/java/org/apache/flink/connector/lance/LanceAggregateSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,10 @@
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;

import com.lancedb.lance.Dataset;
import com.lancedb.lance.Fragment;
import com.lancedb.lance.ipc.LanceScanner;
import com.lancedb.lance.ipc.ScanOptions;
import org.lance.Dataset;
import org.lance.Fragment;
import org.lance.ipc.LanceScanner;
import org.lance.ipc.ScanOptions;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
Expand Down Expand Up @@ -111,7 +111,7 @@ public void open(Configuration parameters) throws Exception {
}

try {
this.dataset = Dataset.open(datasetPath, allocator);
this.dataset = Dataset.open().allocator(allocator).uri(datasetPath).build();
} catch (Exception e) {
throw new IOException("Failed to open Lance dataset: " + datasetPath, e);
}
Expand Down
42 changes: 20 additions & 22 deletions src/main/java/org/apache/flink/connector/lance/LanceIndexBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,15 @@

import org.apache.flink.connector.lance.config.LanceOptions;

import com.lancedb.lance.Dataset;
import com.lancedb.lance.index.DistanceType;
import com.lancedb.lance.index.IndexParams;
import com.lancedb.lance.index.IndexType;
import com.lancedb.lance.index.vector.HnswBuildParams;
import com.lancedb.lance.index.vector.IvfBuildParams;
import com.lancedb.lance.index.vector.PQBuildParams;
import com.lancedb.lance.index.vector.VectorIndexParams;
import org.lance.Dataset;
import org.lance.index.DistanceType;
import org.lance.index.IndexOptions;
import org.lance.index.IndexParams;
import org.lance.index.IndexType;
import org.lance.index.vector.HnswBuildParams;
import org.lance.index.vector.IvfBuildParams;
import org.lance.index.vector.PQBuildParams;
import org.lance.index.vector.VectorIndexParams;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.slf4j.Logger;
Expand Down Expand Up @@ -105,7 +106,7 @@ public IndexBuildResult buildIndex() throws IOException {
try {
// Initialize resources
this.allocator = new RootAllocator(Long.MAX_VALUE);
this.dataset = Dataset.open(datasetPath, allocator);
this.dataset = Dataset.open().allocator(allocator).uri(datasetPath).build();

// Validate column exists
validateColumn();
Expand All @@ -131,8 +132,7 @@ public IndexBuildResult buildIndex() throws IOException {
.build();
VectorIndexParams ivfPqParams = VectorIndexParams.withIvfPqParams(
distanceType, ivfParams, pqParams);
indexParams = new IndexParams.Builder()
.setDistanceType(distanceType)
indexParams = IndexParams.builder()
.setVectorIndexParams(ivfPqParams)
.build();
break;
Expand All @@ -150,17 +150,15 @@ public IndexBuildResult buildIndex() throws IOException {
.build();
VectorIndexParams ivfHnswParams = VectorIndexParams.withIvfHnswPqParams(
distanceType, ivfParams, hnswParams, hnswPqParams);
indexParams = new IndexParams.Builder()
.setDistanceType(distanceType)
indexParams = IndexParams.builder()
.setVectorIndexParams(ivfHnswParams)
.build();
break;

case IVF_FLAT:
lanceIndexType = IndexType.IVF_FLAT;
VectorIndexParams ivfFlatParams = VectorIndexParams.ivfFlat(numPartitions, distanceType);
indexParams = new IndexParams.Builder()
.setDistanceType(distanceType)
indexParams = IndexParams.builder()
.setVectorIndexParams(ivfFlatParams)
.build();
break;
Expand All @@ -170,13 +168,13 @@ public IndexBuildResult buildIndex() throws IOException {
}

// Create index
dataset.createIndex(
Collections.singletonList(columnName),
lanceIndexType,
Optional.empty(), // Index name, use default
indexParams,
replace
);
IndexOptions indexOptions = IndexOptions
.builder(Collections.singletonList(columnName), lanceIndexType, indexParams)
.replace(replace)
.withIndexName(null)
.build();

dataset.createIndex(indexOptions);

long endTime = System.currentTimeMillis();
long duration = endTime - startTime;
Expand Down
12 changes: 6 additions & 6 deletions src/main/java/org/apache/flink/connector/lance/LanceInputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,10 @@
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;

import com.lancedb.lance.Dataset;
import com.lancedb.lance.Fragment;
import com.lancedb.lance.ipc.LanceScanner;
import com.lancedb.lance.ipc.ScanOptions;
import org.lance.Dataset;
import org.lance.Fragment;
import org.lance.ipc.LanceScanner;
import org.lance.ipc.ScanOptions;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
Expand Down Expand Up @@ -107,7 +107,7 @@ public LanceSplit[] createInputSplits(int minNumSplits) throws IOException {

BufferAllocator tempAllocator = new RootAllocator(Long.MAX_VALUE);
try {
Dataset tempDataset = Dataset.open(datasetPath, tempAllocator);
Dataset tempDataset = Dataset.open().allocator(tempAllocator).uri(datasetPath).build();
try {
List<Fragment> fragments = tempDataset.getFragments();
LanceSplit[] splits = new LanceSplit[fragments.size()];
Expand Down Expand Up @@ -143,7 +143,7 @@ public void open(LanceSplit split) throws IOException {
// Open dataset
String datasetPath = split.getDatasetPath();
try {
this.dataset = Dataset.open(datasetPath, allocator);
this.dataset = Dataset.open().allocator(allocator).uri(datasetPath).build();
} catch (Exception e) {
throw new IOException("Cannot open dataset: " + datasetPath, e);
}
Expand Down
51 changes: 32 additions & 19 deletions src/main/java/org/apache/flink/connector/lance/LanceSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,15 +29,18 @@
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;

import com.lancedb.lance.Dataset;
import com.lancedb.lance.Fragment;
import com.lancedb.lance.FragmentMetadata;
import com.lancedb.lance.FragmentOperation;
import com.lancedb.lance.WriteParams;
import org.lance.Dataset;
import org.lance.Fragment;
import org.lance.FragmentMetadata;
import org.lance.WriteParams;
import org.lance.CommitBuilder;
import org.lance.Transaction;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.types.pojo.Schema;
import org.lance.operation.Append;
import org.lance.operation.Overwrite;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -159,33 +162,43 @@ public void flush() throws IOException {
WriteParams writeParams = new WriteParams.Builder()
.withMaxRowsPerFile(options.getWriteMaxRowsPerFile())
.build();

// Create Fragment
List<FragmentMetadata> fragments = Fragment.create(
datasetPath,
allocator,
root,
writeParams
);
List<FragmentMetadata> fragments = Fragment.write()
.datasetUri(datasetPath)
.allocator(allocator)
.data(root)
.writeParams(writeParams)
.execute();

if (!datasetExists) {
// Create new dataset (using Overwrite operation)
FragmentOperation.Overwrite overwrite = new FragmentOperation.Overwrite(fragments, arrowSchema);
dataset = overwrite.commit(allocator, datasetPath, Optional.empty(), Collections.emptyMap());
Overwrite operation = Overwrite.builder().fragments(fragments).schema(arrowSchema).build();
final CommitBuilder builder =
new CommitBuilder(datasetPath, allocator).writeParams(Collections.emptyMap());
try (Transaction txn = new Transaction.Builder().operation(operation).build()) {
dataset = builder.execute(txn);
}
datasetExists = true;
isFirstWrite = false;
LOG.info("Created new dataset: {}", datasetPath);
} else {
// Append data
if (isFirstWrite && options.getWriteMode() == LanceOptions.WriteMode.OVERWRITE) {
// First write and overwrite mode
FragmentOperation.Overwrite overwrite = new FragmentOperation.Overwrite(fragments, arrowSchema);
dataset = overwrite.commit(allocator, datasetPath, Optional.empty(), Collections.emptyMap());
Overwrite operation = Overwrite.builder().fragments(fragments).schema(arrowSchema).build();
final CommitBuilder builder = new CommitBuilder(datasetPath, allocator);
try (Transaction txn = new Transaction.Builder().operation(operation).build()) {
dataset = builder.execute(txn);
}
isFirstWrite = false;
} else {
// Append mode
FragmentOperation.Append append = new FragmentOperation.Append(fragments);
dataset = append.commit(allocator, datasetPath, Optional.empty(), Collections.emptyMap());
Append operation = Append.builder().fragments(fragments).build();
final CommitBuilder builder = new CommitBuilder(datasetPath, allocator);
try (Transaction txn = new Transaction.Builder().operation(operation).build()) {
dataset = builder.execute(txn);
}
}
}

Expand Down
10 changes: 5 additions & 5 deletions src/main/java/org/apache/flink/connector/lance/LanceSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,10 @@
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;

import com.lancedb.lance.Dataset;
import com.lancedb.lance.Fragment;
import com.lancedb.lance.ipc.LanceScanner;
import com.lancedb.lance.ipc.ScanOptions;
import org.lance.Dataset;
import org.lance.Fragment;
import org.lance.ipc.LanceScanner;
import org.lance.ipc.ScanOptions;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
Expand Down Expand Up @@ -126,7 +126,7 @@ public void open(Configuration parameters) throws Exception {

Path path = Paths.get(datasetPath);
try {
this.dataset = Dataset.open(path.toString(), allocator);
this.dataset = Dataset.open().allocator(allocator).uri(path.toString()).build();
} catch (Exception e) {
throw new IOException("Cannot open Lance dataset: " + datasetPath, e);
}
Expand Down
12 changes: 6 additions & 6 deletions src/main/java/org/apache/flink/connector/lance/LanceVectorSearch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,11 @@
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;

import com.lancedb.lance.Dataset;
import com.lancedb.lance.index.DistanceType;
import com.lancedb.lance.ipc.LanceScanner;
import com.lancedb.lance.ipc.Query;
import com.lancedb.lance.ipc.ScanOptions;
import org.lance.Dataset;
import org.lance.index.DistanceType;
import org.lance.ipc.LanceScanner;
import org.lance.ipc.Query;
import org.lance.ipc.ScanOptions;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.Float8Vector;
Expand Down Expand Up @@ -99,7 +99,7 @@ public void open() throws IOException {
this.allocator = new RootAllocator(Long.MAX_VALUE);

try {
this.dataset = Dataset.open(datasetPath, allocator);
this.dataset = Dataset.open().allocator(allocator).uri(datasetPath).build();

// Get Schema and create converter
Schema arrowSchema = dataset.getSchema();
Expand Down
6 changes: 3 additions & 3 deletions src/main/java/org/apache/flink/connector/lance/table/LanceCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;

import com.lancedb.lance.Dataset;
import org.lance.Dataset;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.slf4j.Logger;
Expand Down Expand Up @@ -421,7 +421,7 @@ public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistExcep
if (isRemoteStorage) {
configureStorageEnvironment();
}
Dataset dataset = Dataset.open(datasetPath, allocator);
Dataset dataset = Dataset.open().allocator(allocator).uri(datasetPath).build();

try {
// Infer Flink Schema from Lance Schema
Expand Down Expand Up @@ -476,7 +476,7 @@ public boolean tableExists(ObjectPath tablePath) throws CatalogException {
// Try to open dataset to verify existence
try {
configureStorageEnvironment();
Dataset dataset = Dataset.open(datasetPath, allocator);
Dataset dataset = Dataset.open().allocator(allocator).uri(datasetPath).build();
dataset.close();
knownTables.add(tableKey);
return true;
Expand Down