Skip to content

fix: use dataset file format version to write data files #280

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
jackye1995 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix: use dataset file format version to write data files #280

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
09fb1c3
feat: allow setting lance file format version
jackye1995 Feb 28, 2026
2a4bf6f
fix test
jackye1995 Feb 28, 2026
70edca0
commit
jackye1995 Feb 28, 2026
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
16 changes: 12 additions & 4 deletions lance-spark-3.4_2.12/src/main/java/org/lance/spark/LanceNamespaceSparkCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,9 +64,15 @@ public LanceDataset createDataset(
StructType sparkSchema,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties) {
Map<String, String> namespaceProperties,
String fileFormatVersion) {
return new LancePositionDeltaDataset(
readOptions, sparkSchema, initialStorageOptions, namespaceImpl, namespaceProperties);
readOptions,
sparkSchema,
initialStorageOptions,
namespaceImpl,
namespaceProperties,
fileFormatVersion);
}

@Override
Expand All @@ -76,13 +82,15 @@ public LanceDataset createStagedDataset(
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties,
StagedCommit stagedCommit) {
StagedCommit stagedCommit,
String fileFormatVersion) {
return new LancePositionDeltaDataset(
readOptions,
sparkSchema,
initialStorageOptions,
namespaceImpl,
namespaceProperties,
stagedCommit);
stagedCommit,
fileFormatVersion);
}
}
38 changes: 37 additions & 1 deletion lance-spark-3.4_2.12/src/main/java/org/lance/spark/LancePositionDeltaDataset.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,41 @@ public LancePositionDeltaDataset(
stagedCommit);
}

public LancePositionDeltaDataset(
LanceSparkReadOptions readOptions,
StructType sparkSchema,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties,
String fileFormatVersion) {
super(
readOptions,
sparkSchema,
initialStorageOptions,
namespaceImpl,
namespaceProperties,
null,
fileFormatVersion);
}

public LancePositionDeltaDataset(
LanceSparkReadOptions readOptions,
StructType sparkSchema,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties,
StagedCommit stagedCommit,
String fileFormatVersion) {
super(
readOptions,
sparkSchema,
initialStorageOptions,
namespaceImpl,
namespaceProperties,
stagedCommit,
fileFormatVersion);
}

@Override
public RowLevelOperationBuilder newRowLevelOperationBuilder(
RowLevelOperationInfo rowLevelOperationInfo) {
Expand All @@ -57,6 +92,7 @@ public RowLevelOperationBuilder newRowLevelOperationBuilder(
readOptions(),
getInitialStorageOptions(),
getNamespaceImpl(),
getNamespaceProperties());
getNamespaceProperties(),
getFileFormatVersion());
}
}
15 changes: 11 additions & 4 deletions lance-spark-3.4_2.12/src/main/java/org/lance/spark/LancePositionDeltaOperation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,19 +44,23 @@ public class LancePositionDeltaOperation implements RowLevelOperation, SupportsD

private final Map<String, String> namespaceProperties;

private final String fileFormatVersion;

public LancePositionDeltaOperation(
Command command,
StructType sparkSchema,
LanceSparkReadOptions readOptions,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties) {
Map<String, String> namespaceProperties,
String fileFormatVersion) {
this.command = command;
this.sparkSchema = sparkSchema;
this.readOptions = readOptions;
this.initialStorageOptions = initialStorageOptions;
this.namespaceImpl = namespaceImpl;
this.namespaceProperties = namespaceProperties;
this.fileFormatVersion = fileFormatVersion;
}

@Override
Expand All @@ -73,13 +77,16 @@ public ScanBuilder newScanBuilder(CaseInsensitiveStringMap caseInsensitiveString
@Override
public DeltaWriteBuilder newWriteBuilder(LogicalWriteInfo logicalWriteInfo) {
// Create write options from read options for delta operations
LanceSparkWriteOptions writeOptions =
LanceSparkWriteOptions.Builder writeOptionsBuilder =
LanceSparkWriteOptions.builder()
.datasetUri(readOptions.getDatasetUri())
.storageOptions(readOptions.getStorageOptions())
.namespace(readOptions.getNamespace())
.tableId(readOptions.getTableId())
.build();
.tableId(readOptions.getTableId());
if (fileFormatVersion != null) {
writeOptionsBuilder.fileFormatVersion(fileFormatVersion);
}
LanceSparkWriteOptions writeOptions = writeOptionsBuilder.build();
return new SparkPositionDeltaWriteBuilder(
sparkSchema,
writeOptions,
Expand Down
9 changes: 7 additions & 2 deletions lance-spark-3.4_2.12/src/main/java/org/lance/spark/LanceRowLevelOperationBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,19 +35,23 @@ public class LanceRowLevelOperationBuilder implements RowLevelOperationBuilder {

private final Map<String, String> namespaceProperties;

private final String fileFormatVersion;

public LanceRowLevelOperationBuilder(
RowLevelOperation.Command command,
StructType sparkSchema,
LanceSparkReadOptions readOptions,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties) {
Map<String, String> namespaceProperties,
String fileFormatVersion) {
this.command = command;
this.sparkSchema = sparkSchema;
this.readOptions = readOptions;
this.initialStorageOptions = initialStorageOptions;
this.namespaceImpl = namespaceImpl;
this.namespaceProperties = namespaceProperties;
this.fileFormatVersion = fileFormatVersion;
}

@Override
Expand All @@ -58,6 +62,7 @@ public RowLevelOperation build() {
readOptions,
initialStorageOptions,
namespaceImpl,
namespaceProperties);
namespaceProperties,
fileFormatVersion);
}
}
4 changes: 2 additions & 2 deletions lance-spark-3.4_2.12/src/main/java/org/lance/spark/write/SparkPositionDeltaWrite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,8 +271,8 @@ private WriteParams buildWriteParams() {
if (writeOptions.getMaxBytesPerFile() != null) {
builder.withMaxBytesPerFile(writeOptions.getMaxBytesPerFile());
}
if (writeOptions.getDataStorageVersion() != null) {
builder.withDataStorageVersion(writeOptions.getDataStorageVersion());
if (writeOptions.getFileFormatVersion() != null) {
builder.withDataStorageVersion(writeOptions.getFileFormatVersion());
}
builder.withStorageOptions(merged);
return builder.build();
Expand Down
16 changes: 12 additions & 4 deletions lance-spark-3.5_2.12/src/main/java/org/lance/spark/LanceNamespaceSparkCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,15 @@ public LanceDataset createDataset(
StructType sparkSchema,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties) {
Map<String, String> namespaceProperties,
String fileFormatVersion) {
return new LancePositionDeltaDataset(
readOptions, sparkSchema, initialStorageOptions, namespaceImpl, namespaceProperties);
readOptions,
sparkSchema,
initialStorageOptions,
namespaceImpl,
namespaceProperties,
fileFormatVersion);
}

@Override
Expand All @@ -39,13 +45,15 @@ public LanceDataset createStagedDataset(
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties,
StagedCommit stagedCommit) {
StagedCommit stagedCommit,
String fileFormatVersion) {
return new LancePositionDeltaDataset(
readOptions,
sparkSchema,
initialStorageOptions,
namespaceImpl,
namespaceProperties,
stagedCommit);
stagedCommit,
fileFormatVersion);
}
}
38 changes: 37 additions & 1 deletion lance-spark-3.5_2.12/src/main/java/org/lance/spark/LancePositionDeltaDataset.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,41 @@ public LancePositionDeltaDataset(
stagedCommit);
}

public LancePositionDeltaDataset(
LanceSparkReadOptions readOptions,
StructType sparkSchema,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties,
String fileFormatVersion) {
super(
readOptions,
sparkSchema,
initialStorageOptions,
namespaceImpl,
namespaceProperties,
null,
fileFormatVersion);
}

public LancePositionDeltaDataset(
LanceSparkReadOptions readOptions,
StructType sparkSchema,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties,
StagedCommit stagedCommit,
String fileFormatVersion) {
super(
readOptions,
sparkSchema,
initialStorageOptions,
namespaceImpl,
namespaceProperties,
stagedCommit,
fileFormatVersion);
}

@Override
public RowLevelOperationBuilder newRowLevelOperationBuilder(
RowLevelOperationInfo rowLevelOperationInfo) {
Expand All @@ -57,6 +92,7 @@ public RowLevelOperationBuilder newRowLevelOperationBuilder(
readOptions(),
getInitialStorageOptions(),
getNamespaceImpl(),
getNamespaceProperties());
getNamespaceProperties(),
getFileFormatVersion());
}
}
15 changes: 11 additions & 4 deletions lance-spark-3.5_2.12/src/main/java/org/lance/spark/LancePositionDeltaOperation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,19 +44,23 @@ public class LancePositionDeltaOperation implements RowLevelOperation, SupportsD

private final Map<String, String> namespaceProperties;

private final String fileFormatVersion;

public LancePositionDeltaOperation(
Command command,
StructType sparkSchema,
LanceSparkReadOptions readOptions,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties) {
Map<String, String> namespaceProperties,
String fileFormatVersion) {
this.command = command;
this.sparkSchema = sparkSchema;
this.readOptions = readOptions;
this.initialStorageOptions = initialStorageOptions;
this.namespaceImpl = namespaceImpl;
this.namespaceProperties = namespaceProperties;
this.fileFormatVersion = fileFormatVersion;
}

@Override
Expand All @@ -73,13 +77,16 @@ public ScanBuilder newScanBuilder(CaseInsensitiveStringMap caseInsensitiveString
@Override
public DeltaWriteBuilder newWriteBuilder(LogicalWriteInfo logicalWriteInfo) {
// Create write options from read options for delta operations
LanceSparkWriteOptions writeOptions =
LanceSparkWriteOptions.Builder writeOptionsBuilder =
LanceSparkWriteOptions.builder()
.datasetUri(readOptions.getDatasetUri())
.storageOptions(readOptions.getStorageOptions())
.namespace(readOptions.getNamespace())
.tableId(readOptions.getTableId())
.build();
.tableId(readOptions.getTableId());
if (fileFormatVersion != null) {
writeOptionsBuilder.fileFormatVersion(fileFormatVersion);
}
LanceSparkWriteOptions writeOptions = writeOptionsBuilder.build();
return new SparkPositionDeltaWriteBuilder(
sparkSchema,
writeOptions,
Expand Down
9 changes: 7 additions & 2 deletions lance-spark-3.5_2.12/src/main/java/org/lance/spark/LanceRowLevelOperationBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,19 +35,23 @@ public class LanceRowLevelOperationBuilder implements RowLevelOperationBuilder {

private final Map<String, String> namespaceProperties;

private final String fileFormatVersion;

public LanceRowLevelOperationBuilder(
RowLevelOperation.Command command,
StructType sparkSchema,
LanceSparkReadOptions readOptions,
Map<String, String> initialStorageOptions,
String namespaceImpl,
Map<String, String> namespaceProperties) {
Map<String, String> namespaceProperties,
String fileFormatVersion) {
this.command = command;
this.sparkSchema = sparkSchema;
this.readOptions = readOptions;
this.initialStorageOptions = initialStorageOptions;
this.namespaceImpl = namespaceImpl;
this.namespaceProperties = namespaceProperties;
this.fileFormatVersion = fileFormatVersion;
}

@Override
Expand All @@ -58,6 +62,7 @@ public RowLevelOperation build() {
readOptions,
initialStorageOptions,
namespaceImpl,
namespaceProperties);
namespaceProperties,
fileFormatVersion);
}
}
4 changes: 2 additions & 2 deletions lance-spark-3.5_2.12/src/main/java/org/lance/spark/write/SparkPositionDeltaWrite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,8 +271,8 @@ private WriteParams buildWriteParams() {
if (writeOptions.getMaxBytesPerFile() != null) {
builder.withMaxBytesPerFile(writeOptions.getMaxBytesPerFile());
}
if (writeOptions.getDataStorageVersion() != null) {
builder.withDataStorageVersion(writeOptions.getDataStorageVersion());
if (writeOptions.getFileFormatVersion() != null) {
builder.withDataStorageVersion(writeOptions.getFileFormatVersion());
}
builder.withStorageOptions(merged);
return builder.build();
Expand Down
Loading