fixup! Add Lance connector

Author: luohao

.github/workflows/ci.yml

@@ -376,6 +376,7 @@ jobs:
             !:trino-jdbc,
             !:trino-kafka,
             !:trino-lakehouse,
+            !:trino-lance,
             !:trino-main,
             !:trino-mariadb,
             !:trino-memory,
@@ -499,6 +500,7 @@ jobs:
             - { modules: plugin/trino-ignite }
             - { modules: plugin/trino-kafka }
             - { modules: plugin/trino-lakehouse }
+            - { modules: plugin/trino-lance }
             - { modules: plugin/trino-mariadb }
             - { modules: plugin/trino-mongodb }
             - { modules: plugin/trino-mysql }
@@ -922,6 +924,7 @@ jobs:
             - suite-storage-formats-detailed
             - suite-parquet
             - suite-oauth2
+            - suite-lance
             - suite-ldap
             - suite-loki
             - suite-compatibility

core/trino-server/src/main/provisio/trino.xml

@@ -157,6 +157,12 @@
         </artifact>
     </artifactSet>
 
+    <artifactSet to="plugin/lance">
+        <artifact id="${project.groupId}:trino-lance:zip:${project.version}">
+            <unpack />
+        </artifact>
+    </artifactSet>
+
     <artifactSet to="plugin/loki">
         <artifact id="${project.groupId}:trino-loki:zip:${project.version}">
             <unpack />

docs/src/main/sphinx/connector.md

@@ -26,6 +26,7 @@ Ignite          <connector/ignite>
 JMX             <connector/jmx>
 Kafka           <connector/kafka>
 Lakehouse       <connector/lakehouse>
+Lance           <connector/lance>
 Loki            <connector/loki>
 MariaDB         <connector/mariadb>
 Memory          <connector/memory>

docs/src/main/sphinx/connector/lance.md

@@ -0,0 +1,46 @@
+# Lance connector
+
+## General configuration
+
+To configure the Lance connector, create a catalog properties file `etc/catalog/example.properties` with the following content
+
+```text
+connector.name=lance
+```
+
+You must configure a [namespace](https://lancedb.github.io/lance/format/namespace/) type.
+Currently only [directory namespace](https://lancedb.github.io/lance/format/namespace/impls/dir/) is supported.
+
+```text
+lance.namespace.type=directory
+```
+
+## Lance Namespace configuration
+### Directory namespace
+Lance directory namespace is a lightweight and simple single-level Lance namespace that contains only a list of tables. All tables reside in the default namespace.
+
+The following configuration properties are available:
+
+:::{list-table}
+:widths: 30, 10, 60
+:header-rows: 1
+
+* - Property name
+  - Required
+  - Description
+* - `lance.namespace.directory.warehouse.location`
+  - Yes
+  - The root directory URI of the namespace where tables are stored.
+:::
+
+
+## File system access configuration
+
+The connector supports accessing the following file systems:
+
+* [](/object-storage/file-system-azure)
+* [](/object-storage/file-system-gcs)
+* [](/object-storage/file-system-s3)
+* [](/object-storage/file-system-hdfs)
+
+You must enable and configure the specific file system access. 

lib/trino-lance-file/pom.xml

@@ -13,6 +13,7 @@
     <description>Trino - Lance file format</description>
 
     <properties>
+        <air.compiler.fail-warnings>true</air.compiler.fail-warnings>
         <air.test.jvm.additional-arguments>${air.test.jvm.additional-arguments.default}
             --add-opens=java.base/java.nio=ALL-UNNAMED
             --sun-misc-unsafe-memory-access=allow</air.test.jvm.additional-arguments>
@@ -24,11 +25,6 @@
             <artifactId>lancedb_lance_protocolbuffers_java</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-annotations</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>com.google.errorprone</groupId>
             <artifactId>error_prone_annotations</artifactId>
@@ -47,6 +43,7 @@
         <dependency>
             <groupId>io.github.luohao</groupId>
             <artifactId>fastlanes-java</artifactId>
+            <version>1</version>
         </dependency>
 
         <dependency>

lib/trino-lance-file/src/main/java/io/trino/lance/file/LanceReader.java

@@ -22,7 +22,7 @@
 import io.trino.lance.file.v2.metadata.Field;
 import io.trino.lance.file.v2.metadata.FileVersion;
 import io.trino.lance.file.v2.metadata.Footer;
-import io.trino.lance.file.v2.metadata.TypeUtil;
+import io.trino.lance.file.v2.metadata.LanceTypeUtil;
 import io.trino.lance.file.v2.reader.ColumnReader;
 import io.trino.lance.file.v2.reader.Range;
 import io.trino.memory.context.AggregatedMemoryContext;
@@ -82,7 +82,7 @@ public LanceReader(LanceDataSource dataSource,
         // read Global Buffer Offset Table
         Slice bufferOffsetTableSlice = dataSource.readFully(footer.getGlobalBuffOffsetStart(), footer.getNumGlobalBuffers() * BUFFER_DESCRIPTOR_SIZE);
         List<DiskRange> bufferOffsets = IntStream.range(0, footer.getNumGlobalBuffers()).boxed()
-                .map(i -> DiskRange.of(bufferOffsetTableSlice.getLong(BUFFER_DESCRIPTOR_SIZE * i), bufferOffsetTableSlice.getLong(BUFFER_DESCRIPTOR_SIZE * i + 8)))
+                .map(i -> new DiskRange(bufferOffsetTableSlice.getLong(BUFFER_DESCRIPTOR_SIZE * i), bufferOffsetTableSlice.getLong(BUFFER_DESCRIPTOR_SIZE * i + 8)))
                 .collect(toImmutableList());
         if (bufferOffsets.size() == 0) {
             throw new RuntimeException("File did not contain any buffers");
@@ -91,40 +91,43 @@ public LanceReader(LanceDataSource dataSource,
         // read global schema
         DiskRange schemaBufferLocation = bufferOffsets.get(0);
         // prefetch all metadata
-        Slice metadataSlice = dataSource.readTail(toIntExact(dataSource.getEstimatedSize() - schemaBufferLocation.getPosition()));
+        Slice metadataSlice = dataSource.readTail(toIntExact(dataSource.getEstimatedSize() - schemaBufferLocation.position()));
         // read file descriptor
-        Slice schemaSlice = metadataSlice.slice(0, toIntExact(schemaBufferLocation.getLength()));
+        Slice schemaSlice = metadataSlice.slice(0, toIntExact(schemaBufferLocation.length()));
         build.buf.gen.lance.file.FileDescriptor fileDescriptor = build.buf.gen.lance.file.FileDescriptor.parseFrom(schemaSlice.toByteBuffer());
         checkArgument(fileDescriptor.hasSchema(), "FileDescriptor does not contain a schema");
         this.fields = toFields(fileDescriptor.getSchema());
-        List<Range> ranges = requestRanges.orElse(ImmutableList.of(Range.of(0, fileDescriptor.getLength())));
+        List<Range> ranges = requestRanges.orElse(ImmutableList.of(new Range(0, fileDescriptor.getLength())));
         this.numRows = ranges.stream()
                 .mapToLong(Range::length)
                 .sum();
         // read Column Metadata Offset Table
-        Slice columnMetadataOffsetsSlice = metadataSlice.slice(toIntExact(footer.getColumnMetadataOffsetsStart() - schemaBufferLocation.getPosition()), toIntExact(footer.getGlobalBuffOffsetStart() - footer.getColumnMetadataOffsetsStart()));
+        Slice columnMetadataOffsetsSlice = metadataSlice.slice(toIntExact(footer.getColumnMetadataOffsetsStart() - schemaBufferLocation.position()), toIntExact(footer.getGlobalBuffOffsetStart() - footer.getColumnMetadataOffsetsStart()));
         List<DiskRange> columnMetadataOffsets = IntStream.range(0, footer.getNumColumns()).boxed()
-                .map(i -> DiskRange.of(columnMetadataOffsetsSlice.getLong(i * BUFFER_DESCRIPTOR_SIZE), columnMetadataOffsetsSlice.getLong(i * BUFFER_DESCRIPTOR_SIZE + 8)))
+                .map(i -> {
+                    long position = columnMetadataOffsetsSlice.getLong(i * BUFFER_DESCRIPTOR_SIZE);
+                    return new DiskRange(position, columnMetadataOffsetsSlice.getLong(i * BUFFER_DESCRIPTOR_SIZE + 8));
+                })
                 .collect(toImmutableList());
 
         // read Column Metadata
-        Slice columnMetadataSlice = metadataSlice.slice(toIntExact(footer.getColumnMetadataStart() - schemaBufferLocation.getPosition()), toIntExact(footer.getColumnMetadataOffsetsStart() - footer.getColumnMetadataStart()));
+        Slice columnMetadataSlice = metadataSlice.slice(toIntExact(footer.getColumnMetadataStart() - schemaBufferLocation.position()), toIntExact(footer.getColumnMetadataOffsetsStart() - footer.getColumnMetadataStart()));
         List<ColumnMetadata> metadata = IntStream.range(0, footer.getNumColumns()).boxed()
                 .map(i -> {
                     DiskRange offset = columnMetadataOffsets.get(i);
-                    Slice message = columnMetadataSlice.slice(toIntExact(offset.getPosition() - footer.getColumnMetadataStart()), toIntExact(offset.getLength()));
+                    Slice message = columnMetadataSlice.slice(toIntExact(offset.position() - footer.getColumnMetadataStart()), toIntExact(offset.length()));
                     return ColumnMetadata.from(i, message);
                 })
                 .collect(toImmutableList());
 
-        Map<Integer, Integer> fieldIdMap = TypeUtil.visit(fields, new TypeUtil.FieldIdToColumnIndexVisitor());
+        Map<Integer, Integer> fieldIdMap = LanceTypeUtil.visit(fields, new LanceTypeUtil.FieldIdToColumnIndexVisitor());
         this.columnMetadata = fieldIdMap.entrySet().stream()
                 .collect(toImmutableMap(
                         Map.Entry::getKey,
                         entry -> metadata.get(entry.getValue())));
 
         this.columnReaders = fields.stream()
-                .filter(field -> columnIds.contains(field.getId()))
+                .filter(field -> columnIds.contains(field.id()))
                 .map(field ->
                         ColumnReader.createColumnReader(
                                 dataSource,
@@ -149,7 +152,7 @@ public static List<Field> toFields(List<build.buf.gen.lance.file.Field> fieldsPr
         }
         List<Field> fields = new ArrayList<>();
         for (Map.Entry<Integer, Field> entry : fieldMap.entrySet()) {
-            int parentId = fieldMap.get(entry.getKey()).getParentId();
+            int parentId = fieldMap.get(entry.getKey()).parentId();
             Field field = entry.getValue();
             if (parentId == -1) {
                 fields.add(field);
@@ -299,7 +302,7 @@ public Block getBlock(int channel)
 
             Block block = blocks[channel];
             if (block == null) {
-                block = columnReaders[channel].read().getBlock();
+                block = columnReaders[channel].read().block();
                 block = selectedPositions.apply(block);
             }
             blocks[channel] = block;

lib/trino-lance-file/src/main/java/io/trino/lance/file/MemoryLanceDataSource.java

@@ -13,18 +13,16 @@
  */
 package io.trino.lance.file.v2.metadata;
 
-import java.util.List;
+import com.google.common.collect.ImmutableList;
 
-import static java.util.Objects.requireNonNull;
+import java.util.List;
 
-public final class AllNullLayout
+public record AllNullLayout(List<RepDefLayer> layers)
         implements PageLayout
 {
-    public final List<RepDefLayer> layers;
-
-    public AllNullLayout(List<RepDefLayer> layers)
+    public AllNullLayout
     {
-        this.layers = requireNonNull(layers, "layers is null");
+        layers = ImmutableList.copyOf(layers);
     }
 
     public static AllNullLayout fromProto(build.buf.gen.lance.encodings21.AllNullLayout proto)

lib/trino-lance-file/src/main/java/io/trino/lance/file/v2/metadata/AllNullLayout.java

@@ -23,19 +23,13 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
-import static java.util.Objects.requireNonNull;
 
-public class ColumnMetadata
+public record ColumnMetadata(int index, List<PageMetadata> pages, List<DiskRange> bufferOffsets)
 {
-    private final int index;
-    private final List<PageMetadata> pages;
-    private final List<DiskRange> bufferOffsets;
-
-    public ColumnMetadata(int index, List<PageMetadata> pages, List<DiskRange> bufferOffsets)
+    public ColumnMetadata
     {
-        this.index = index;
-        this.pages = requireNonNull(pages, "pages is null");
-        this.bufferOffsets = requireNonNull(bufferOffsets, "bufferOffsets is null");
+        pages = ImmutableList.copyOf(pages);
+        bufferOffsets = ImmutableList.copyOf(bufferOffsets);
     }
 
     public static ColumnMetadata from(int columnIndex, Slice data)
@@ -56,15 +50,21 @@ public static ColumnMetadata from(int columnIndex, Slice data)
                     long priority = page.getPriority();
                     int bufferCount = page.getBufferOffsetsList().size();
                     List<DiskRange> buffers = IntStream.range(0, bufferCount).boxed()
-                            .map(i -> DiskRange.of(page.getBufferOffsets(i), page.getBufferSizes(i)))
+                            .map(i -> {
+                                long position = page.getBufferOffsets(i);
+                                return new DiskRange(position, page.getBufferSizes(i));
+                            })
                             .collect(toImmutableList());
                     return new PageMetadata(numRows, priority, getPageLayout(page), buffers);
                 })
                 .collect(toImmutableList());
 
         int bufferCount = proto.getBufferOffsetsList().size();
-        ImmutableList<DiskRange> buffers = IntStream.range(0, bufferCount).boxed()
-                .map(index -> DiskRange.of(proto.getBufferOffsets(index), proto.getBufferSizes(index)))
+        List<DiskRange> buffers = IntStream.range(0, bufferCount).boxed()
+                .map(index -> {
+                    long position = proto.getBufferOffsets(index);
+                    return new DiskRange(position, proto.getBufferSizes(index));
+                })
                 .collect(toImmutableList());
         return new ColumnMetadata(columnIndex, pages, buffers);
     }
@@ -88,19 +88,4 @@ private static PageLayout getPageLayout(build.buf.gen.lance.file.v2.ColumnMetada
             default -> throw new UnsupportedOperationException("Invalid encoding: " + encoding);
         };
     }
-
-    public int getIndex()
-    {
-        return index;
-    }
-
-    public List<PageMetadata> getPages()
-    {
-        return pages;
-    }
-
-    public List<DiskRange> getBufferOffsets()
-    {
-        return bufferOffsets;
-    }
 }