Access OSS Tables through the Java SDK

Create a database, write, and read

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.BigIntVector;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowStreamReader;
import org.apache.arrow.vector.ipc.ArrowStreamWriter;
import org.apache.arrow.vector.types.pojo.*;
import org.lance.Dataset;
import org.lance.ReadOptions;
import org.lance.WriteParams;
import org.lance.namespace.model.*;

List<String> tableId = Arrays.asList("my_db", "my_table");

// Create the namespace
ns.createNamespace(new CreateNamespaceRequest().id(Collections.singletonList("my_db")));

// Reuse the allocator from Step 2; build Arrow data and serialize it to an IPC stream byte array
Schema schema = new Schema(Arrays.asList(
    new Field("id", FieldType.nullable(new ArrowType.Int(64, true)), null),
    new Field("name", FieldType.nullable(new ArrowType.Utf8()), null)));

byte[] ipc;

try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator);
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     ArrowStreamWriter w = new ArrowStreamWriter(root, null, out)) {
  BigIntVector id = (BigIntVector) root.getVector("id");
  VarCharVector name = (VarCharVector) root.getVector("name");
  id.allocateNew(3);
  name.allocateNew(3);
  for (int i = 0; i < 3; i++) {
    id.setSafe(i, i + 1);
    name.setSafe(i, ("row-" + (i + 1)).getBytes(StandardCharsets.UTF_8));
  }
  root.setRowCount(3);
  w.start();
  w.writeBatch();
  w.end();
  ipc = out.toByteArray();
}

// Write (reader is any ArrowReader)
try (ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(ipc), allocator)) {
  Dataset.write()
      .allocator(allocator).reader(reader)
      .namespaceClient(ns).tableId(tableId)
      .mode(WriteParams.WriteMode.CREATE)
      .storageOptions(storageOptions)
      .execute().close();
}

// Read
try (Dataset ds = Dataset.open()
    .allocator(allocator)
    .namespaceClient(ns).tableId(tableId)
    .readOptions(new ReadOptions.Builder().setStorageOptions(storageOptions).build())
    .build()) {
  System.out.println(ds.countRows());   // Outputs 3
}

WriteParams.WriteMode supports modes such as CREATE (create new), APPEND (append), and OVERWRITE (overwrite).

Row-level DML (update / delete / mergeInsert)

Row-level modifications are performed on an open Dataset. The following example reuses tableId, ipc, ns, allocator, and storageOptions from above, and closes the dataset after the operations.

import java.util.Map;
import org.apache.arrow.c.ArrowArrayStream;
import org.apache.arrow.c.Data;
import org.lance.merge.MergeInsertParams;
import org.lance.update.UpdateParams;

try (Dataset ds = Dataset.open()
    .allocator(allocator)
    .namespaceClient(ns).tableId(tableId)
    .readOptions(new ReadOptions.Builder().setStorageOptions(storageOptions).build())
    .build()) {
  ds.delete("id = 3");

  long updated = ds.update(
      new UpdateParams(Map.of("name", "'updated'")).withWhere("id = 1"))
      .getNumRowsUpdated();

  try (ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(ipc), allocator);
       ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator)) {
    Data.exportArrayStream(allocator, reader, stream);
    MergeInsertParams p = new MergeInsertParams(Collections.singletonList("id"))
        .withMatchedUpdateAll()
        .withNotMatched(MergeInsertParams.WhenNotMatched.InsertAll);
    ds.mergeInsert(p, stream);
  }
}

The values in UpdateParams are SQL expressions; string literals must keep their inner single quotes.

catalog operations

The request/response models are all under the org.lance.namespace.model package, and the method names correspond one-to-one with Python (camelCase):

// namespace
ns.namespaceExists(new NamespaceExistsRequest().id(Collections.singletonList("my_db")));
ns.describeNamespace(new DescribeNamespaceRequest().id(Collections.singletonList("my_db")));
ns.listNamespaces(new ListNamespacesRequest().id(Collections.emptyList()));   // An empty id indicates the root

// table
ns.tableExists(new TableExistsRequest().id(tableId));
ns.listTables(new ListTablesRequest().id(Collections.singletonList("my_db")));  // Returns bare table names
ns.listAllTables(new ListTablesRequest().id(Collections.emptyList()));          // Across all namespaces
ns.describeTable(new DescribeTableRequest().id(tableId).loadDetailedMetadata(true)); // Returns version/schema/stats

The optional parameters of describeTable are the same as in Python: checkDeclared(true), withTableUri(true), and loadDetailedMetadata(true). You can also reuse the ipc generated by ArrowStreamWriter above to create another table through the catalog (the server writes the data):

ns.createTable(new CreateTableRequest().id(Arrays.asList("my_db", "t2")), ipc);

Create an index

Dataset.createIndex(IndexOptions) creates scalar and vector indexes; build the parameters with IndexOptions.builder(columns, indexType, params) and specify the index name with withIndexName(...). Scalar index parameters are under the org.lance.index.scalar package, and vector index parameters are under the org.lance.index.vector package.

The my_table from above has an id column, so you can directly create a BTREE index:

import java.util.Collections;
import org.lance.index.IndexOptions;
import org.lance.index.IndexParams;
import org.lance.index.IndexType;
import org.lance.index.scalar.BTreeIndexParams;

try (Dataset ds = Dataset.open()
    .allocator(allocator).namespaceClient(ns).tableId(tableId)
    .readOptions(new ReadOptions.Builder().setStorageOptions(storageOptions).build())
    .build()) {
  IndexParams btreeParams = IndexParams.builder()
      .setScalarIndexParams(BTreeIndexParams.builder().build()).build();
  ds.createIndex(IndexOptions.builder(Collections.singletonList("id"), IndexType.BTREE, btreeParams)
      .withIndexName("id_btree").build());
}

The following full-text and vector examples use another table, search_table, where body is Utf8 and vector is a 64-dimensional FixedSizeList<Float32>. IVF_PQ training requires enough non-empty vectors; this example uses 70,000 rows. First, prepare the sample data:

import org.apache.arrow.vector.Float4Vector;
import org.apache.arrow.vector.complex.FixedSizeListVector;
import org.apache.arrow.vector.types.FloatingPointPrecision;

int dimension = 64;
int rowCount = 70000;
Field item = new Field("item",
    FieldType.nullable(new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE)), null);
Field vectorField = new Field("vector", FieldType.nullable(new ArrowType.FixedSizeList(dimension)),
    Collections.singletonList(item));
Schema searchSchema = new Schema(Arrays.asList(
    new Field("body", FieldType.nullable(new ArrowType.Utf8()), null), vectorField));

byte[] searchIpc;

try (VectorSchemaRoot root = VectorSchemaRoot.create(searchSchema, allocator);
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     ArrowStreamWriter writer = new ArrowStreamWriter(root, null, out)) {
  VarCharVector body = (VarCharVector) root.getVector("body");
  FixedSizeListVector vectors = (FixedSizeListVector) root.getVector("vector");
  body.allocateNew();
  vectors.setInitialCapacity(rowCount);
  vectors.allocateNew();
  Float4Vector values = (Float4Vector) vectors.getDataVector();
  for (int row = 0; row < rowCount; row++) {
    body.setSafe(row, ("the quick brown fox " + row).getBytes(StandardCharsets.UTF_8));
    vectors.setNotNull(row);
    for (int col = 0; col < dimension; col++) {
      values.setSafe(row * dimension + col, ((row * 31 + col * 17) % 1000) / 1000f);
    }
  }
  body.setValueCount(rowCount);
  vectors.setValueCount(rowCount);
  values.setValueCount(rowCount * dimension);
  root.setRowCount(rowCount);
  writer.start();
  writer.writeBatch();
  writer.end();
  searchIpc = out.toByteArray();
}

List<String> searchTableId = Arrays.asList("my_db", "search_table");
try (ArrowStreamReader reader = new ArrowStreamReader(
    new ByteArrayInputStream(searchIpc), allocator)) {
  Dataset.write().allocator(allocator).reader(reader)
      .namespaceClient(ns).tableId(searchTableId)
      .mode(WriteParams.WriteMode.CREATE)
      .storageOptions(storageOptions).execute().close();
}

Create the indexes:

import org.lance.index.DistanceType;
import org.lance.index.scalar.InvertedIndexParams;
import org.lance.index.vector.VectorIndexParams;

try (Dataset searchDs = Dataset.open()
    .allocator(allocator).namespaceClient(ns).tableId(searchTableId)
    .readOptions(new ReadOptions.Builder().setStorageOptions(storageOptions).build())
    .build()) {
  IndexParams invertedParams = IndexParams.builder().setScalarIndexParams(
      InvertedIndexParams.builder().baseTokenizer("simple").language("English").build()).build();
  searchDs.createIndex(IndexOptions.builder(Collections.singletonList("body"), IndexType.INVERTED,
      invertedParams).withIndexName("body_inv").build());

  IndexParams vectorParams = IndexParams.builder()
      .setVectorIndexParams(VectorIndexParams.ivfPq(4, 8, 8, DistanceType.L2, 256))
      .build();
  searchDs.createIndex(IndexOptions.builder(Collections.singletonList("vector"), IndexType.IVF_PQ,
      vectorParams).withIndexName("vec_idx").build());
}

The supported scalar index types are the IndexType enum: BTREE, BITMAP, INVERTED, NGRAM, and ZONEMAP.

Search (vector search and full-text search)

After the indexes are created, use Dataset.newScan(ScanOptions) to obtain a LanceScanner, then call scanBatches() to get an ArrowReader and read the results batch by batch. The following example reuses the search_table from above, which contains body and a 64-dimensional vector.

import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowReader;
import org.lance.index.DistanceType;
import org.lance.ipc.FullTextQuery;
import org.lance.ipc.LanceScanner;
import org.lance.ipc.Query;
import org.lance.ipc.ScanOptions;

try (Dataset searchDs = Dataset.open()
    .allocator(allocator).namespaceClient(ns).tableId(searchTableId)
    .readOptions(new ReadOptions.Builder().setStorageOptions(storageOptions).build())
    .build()) {

  float[] key = new float[dimension];

  Query query = new Query.Builder()
      .setColumn("vector").setKey(key).setK(5)
      .setDistanceType(DistanceType.L2).build();
  ScanOptions vectorOptions = new ScanOptions.Builder().nearest(query).limit(5).build();
  try (LanceScanner scanner = searchDs.newScan(vectorOptions);
       ArrowReader reader = scanner.scanBatches()) {
    while (reader.loadNextBatch()) {
      VectorSchemaRoot root = reader.getVectorSchemaRoot();
      System.out.println("vector hits=" + root.getRowCount());
    }
  }

  FullTextQuery fullTextQuery = FullTextQuery.match("fox", "body");
  ScanOptions textOptions = new ScanOptions.Builder()
      .fullTextQuery(fullTextQuery).limit(10).build();
  try (LanceScanner scanner = searchDs.newScan(textOptions);
       ArrowReader reader = scanner.scanBatches()) {
    int rows = 0;
    while (reader.loadNextBatch()) {
      rows += reader.getVectorSchemaRoot().getRowCount();
    }
    System.out.println("fts hits=" + rows);
  }
}

The parameter order of FullTextQuery.match is match(queryText, column); the vector search result includes a _distance column.

compaction (small file merge)

Multiple writes produce multiple fragments (data files) within a table; Dataset.compact() merges them:

import org.lance.Dataset;

try (Dataset ds = Dataset.open()
    .allocator(allocator).namespaceClient(ns).tableId(tableId)
    .readOptions(new ReadOptions.Builder().setStorageOptions(storageOptions).build())
    .build()) {
  System.out.println(ds.getFragments().size());   // Number of fragments before compaction
  ds.compact();
}

To control the target fragment size, use org.lance.compaction.CompactionOptions to build the parameters and then call ds.compact(options); the default parameters are suitable for routine small file merges.

Rename, deregister, and delete

The following operations change the catalog or delete data; choose the one that matches your intent:

ns.renameTable(new RenameTableRequest().id(tableId).newTableName("my_table_new"));
List<String> renamedTableId = Arrays.asList("my_db", "my_table_new");

Deregister only the catalog metadata and keep the underlying data:

ns.deregisterTable(new DeregisterTableRequest().id(renamedTableId));

Or delete the data:

DropTableResponse dropped = ns.dropTable(new DropTableRequest().id(renamedTableId));

Before deleting a namespace, delete the tables in it first:

ns.dropTable(new DropTableRequest().id(Arrays.asList("my_db", "t2")));
ns.dropTable(new DropTableRequest().id(searchTableId));
ns.dropNamespace(new DropNamespaceRequest().id(Collections.singletonList("my_db")));