This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 936f534c [elixir] feat: Add get_table_info admin call (#640)
936f534c is described below
commit 936f534c3e65a0f9434c801fbf9485f97543d78a
Author: Nicoleta Lazar <[email protected]>
AuthorDate: Fri Jun 26 13:00:23 2026 +0300
[elixir] feat: Add get_table_info admin call (#640)
* [elixir] feat: Add get_table_info admin call
* Include schema in table_info and add get_table_schema admin call
---
bindings/elixir/lib/fluss/admin.ex | 32 +++++
bindings/elixir/lib/fluss/native.ex | 6 +
bindings/elixir/lib/fluss/schema_info.ex | 33 +++++
bindings/elixir/lib/fluss/table_info.ex | 96 ++++++++++++++
bindings/elixir/native/fluss_nif/src/admin.rs | 104 ++++++++++++++-
bindings/elixir/native/fluss_nif/src/schema.rs | 48 +++++++
bindings/elixir/test/integration/admin_test.exs | 165 ++++++++++++++++++++++++
7 files changed, 482 insertions(+), 2 deletions(-)
diff --git a/bindings/elixir/lib/fluss/admin.ex
b/bindings/elixir/lib/fluss/admin.ex
index b9d26f70..11bc2ce8 100644
--- a/bindings/elixir/lib/fluss/admin.ex
+++ b/bindings/elixir/lib/fluss/admin.ex
@@ -176,4 +176,36 @@ defmodule Fluss.Admin do
{:error, %Fluss.Error{} = err} -> raise err
end
end
+
+ @spec get_table_info(t(), String.t(), String.t()) ::
+ {:ok, Fluss.TableInfo.t()} | {:error, Fluss.Error.t()}
+ def get_table_info(admin, database_name, table_name) do
+ admin
+ |> Native.admin_get_table_info(database_name, table_name)
+ |> Native.await_nif()
+ end
+
+ @spec get_table_info!(t(), String.t(), String.t()) :: Fluss.TableInfo.t()
+ def get_table_info!(admin, database_name, table_name) do
+ case get_table_info(admin, database_name, table_name) do
+ {:ok, table_info} -> table_info
+ {:error, %Fluss.Error{} = err} -> raise err
+ end
+ end
+
+ @spec get_table_schema(t(), String.t(), String.t(), integer() | nil) ::
+ {:ok, Fluss.SchemaInfo.t()} | {:error, Fluss.Error.t()}
+ def get_table_schema(admin, database_name, table_name, schema_id \\ nil) do
+ admin
+ |> Native.admin_get_table_schema(database_name, table_name, schema_id)
+ |> Native.await_nif()
+ end
+
+ @spec get_table_schema!(t(), String.t(), String.t(), integer() | nil) ::
Fluss.SchemaInfo.t()
+ def get_table_schema!(admin, database_name, table_name, schema_id \\ nil) do
+ case get_table_schema(admin, database_name, table_name, schema_id) do
+ {:ok, schema_info} -> schema_info
+ {:error, %Fluss.Error{} = err} -> raise err
+ end
+ end
end
diff --git a/bindings/elixir/lib/fluss/native.ex
b/bindings/elixir/lib/fluss/native.ex
index 91b4fbf1..6de93683 100644
--- a/bindings/elixir/lib/fluss/native.ex
+++ b/bindings/elixir/lib/fluss/native.ex
@@ -54,6 +54,12 @@ defmodule Fluss.Native do
def admin_table_exists(_admin, _database_name, _table_name),
do: :erlang.nif_error(:nif_not_loaded)
+ def admin_get_table_info(_admin, _database_name, _table_name),
+ do: :erlang.nif_error(:nif_not_loaded)
+
+ def admin_get_table_schema(_admin, _database_name, _table_name, _schema_id),
+ do: :erlang.nif_error(:nif_not_loaded)
+
# Schema / TableDescriptor
def table_descriptor_new(_schema, _bucket_count, _properties),
do: :erlang.nif_error(:nif_not_loaded)
diff --git a/bindings/elixir/lib/fluss/schema_info.ex
b/bindings/elixir/lib/fluss/schema_info.ex
new file mode 100644
index 00000000..2687ad9f
--- /dev/null
+++ b/bindings/elixir/lib/fluss/schema_info.ex
@@ -0,0 +1,33 @@
+# 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.
+
+defmodule Fluss.SchemaInfo do
+ @moduledoc """
+ A versioned schema: a table's `Fluss.Schema` together with its schema id.
+
+ Returned by `Fluss.Admin.get_table_schema/4`. Fluss versions schemas, so the
+ `:schema_id` identifies which version this is.
+ """
+
+ @enforce_keys [:schema, :schema_id]
+ defstruct [:schema, :schema_id]
+
+ @type t :: %__MODULE__{
+ schema: Fluss.Schema.t(),
+ schema_id: integer()
+ }
+end
diff --git a/bindings/elixir/lib/fluss/table_info.ex
b/bindings/elixir/lib/fluss/table_info.ex
new file mode 100644
index 00000000..31455c1b
--- /dev/null
+++ b/bindings/elixir/lib/fluss/table_info.ex
@@ -0,0 +1,96 @@
+# 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.
+
+defmodule Fluss.TableInfo do
+ @moduledoc """
+ Metadata about a Fluss table: identity, schema id, key/partition layout,
+ bucketing, properties, and creation/modification timestamps as tracked by the
+ cluster.
+
+ Returned by `Fluss.Admin.get_table_info/3`. This carries metadata only — for
a
+ handle to read from or write to the table, use `Fluss.Table.get/3`.
+ """
+
+ @enforce_keys [
+ :database_name,
+ :table_name,
+ :table_id,
+ :schema_id,
+ :schema,
+ :num_buckets,
+ :has_primary_key,
+ :primary_keys,
+ :physical_primary_keys,
+ :bucket_keys,
+ :has_bucket_key,
+ :is_default_bucket_key,
+ :is_partitioned,
+ :is_auto_partitioned,
+ :partition_keys,
+ :comment,
+ :properties,
+ :custom_properties,
+ :created_time,
+ :modified_time
+ ]
+
+ defstruct [
+ :database_name,
+ :table_name,
+ :table_id,
+ :schema_id,
+ :schema,
+ :num_buckets,
+ :has_primary_key,
+ :primary_keys,
+ :physical_primary_keys,
+ :bucket_keys,
+ :has_bucket_key,
+ :is_default_bucket_key,
+ :is_partitioned,
+ :is_auto_partitioned,
+ :partition_keys,
+ :comment,
+ :properties,
+ :custom_properties,
+ :created_time,
+ :modified_time
+ ]
+
+ @type t :: %__MODULE__{
+ database_name: String.t(),
+ table_name: String.t(),
+ table_id: integer(),
+ schema_id: integer(),
+ schema: Fluss.Schema.t(),
+ num_buckets: integer(),
+ has_primary_key: boolean(),
+ primary_keys: [String.t()],
+ physical_primary_keys: [String.t()],
+ bucket_keys: [String.t()],
+ has_bucket_key: boolean(),
+ is_default_bucket_key: boolean(),
+ is_partitioned: boolean(),
+ is_auto_partitioned: boolean(),
+ partition_keys: [String.t()],
+ comment: String.t() | nil,
+ properties: %{String.t() => String.t()},
+ custom_properties: %{String.t() => String.t()},
+ created_time: integer(),
+ modified_time: integer()
+ }
+end
diff --git a/bindings/elixir/native/fluss_nif/src/admin.rs
b/bindings/elixir/native/fluss_nif/src/admin.rs
index 6f79a97f..4d348fc9 100644
--- a/bindings/elixir/native/fluss_nif/src/admin.rs
+++ b/bindings/elixir/native/fluss_nif/src/admin.rs
@@ -18,9 +18,10 @@
use crate::async_nif;
use crate::atoms::to_nif_err;
use crate::connection::ConnectionResource;
-use crate::schema::TableDescriptorResource;
+use crate::schema::{NifSchema, TableDescriptorResource};
use fluss::client::FlussAdmin;
-use fluss::metadata::{DatabaseDescriptor, DatabaseInfo, TablePath};
+use fluss::error::Error;
+use fluss::metadata::{DatabaseDescriptor, DatabaseInfo, SchemaInfo, TableInfo,
TablePath};
use fluss::{ServerNode, ServerType};
use rustler::{Env, NifStruct, NifUnitEnum, ResourceArc, Term};
use std::collections::HashMap;
@@ -105,6 +106,60 @@ impl NifDatabaseInfo {
}
}
+#[derive(NifStruct)]
+#[module = "Fluss.TableInfo"]
+pub struct NifTableInfo {
+ pub database_name: String,
+ pub table_name: String,
+ pub table_id: i64,
+ pub schema_id: i32,
+ pub schema: NifSchema,
+ pub num_buckets: i32,
+ pub has_primary_key: bool,
+ pub primary_keys: Vec<String>,
+ pub physical_primary_keys: Vec<String>,
+ pub bucket_keys: Vec<String>,
+ pub has_bucket_key: bool,
+ pub is_default_bucket_key: bool,
+ pub is_partitioned: bool,
+ pub is_auto_partitioned: bool,
+ pub partition_keys: Vec<String>,
+ pub comment: Option<String>,
+ pub properties: HashMap<String, String>,
+ pub custom_properties: HashMap<String, String>,
+ pub created_time: i64,
+ pub modified_time: i64,
+}
+
+impl NifTableInfo {
+ pub fn from_core(info: &TableInfo) -> Result<Self, Error> {
+ let table_path: &TablePath = info.get_table_path();
+
+ Ok(Self {
+ database_name: table_path.database().to_string(),
+ table_name: table_path.table().to_string(),
+ table_id: info.get_table_id(),
+ schema_id: info.get_schema_id(),
+ schema: NifSchema::from_core(info.get_schema())?,
+ num_buckets: info.get_num_buckets(),
+ has_primary_key: info.has_primary_key(),
+ primary_keys: info.get_primary_keys().to_vec(),
+ physical_primary_keys: info.get_physical_primary_keys().to_vec(),
+ bucket_keys: info.get_bucket_keys().to_vec(),
+ has_bucket_key: info.has_bucket_key(),
+ is_default_bucket_key: info.is_default_bucket_key(),
+ is_partitioned: info.is_partitioned(),
+ is_auto_partitioned: info.is_auto_partitioned(),
+ partition_keys: info.get_partition_keys().to_vec(),
+ comment: info.get_comment().map(String::from),
+ properties: info.get_properties().clone(),
+ custom_properties: info.get_custom_properties().clone(),
+ created_time: info.get_created_time(),
+ modified_time: info.get_modified_time(),
+ })
+ }
+}
+
pub struct AdminResource {
pub inner: Arc<FlussAdmin>,
}
@@ -247,3 +302,48 @@ fn admin_table_exists<'a>(
admin.inner.table_exists(&table_path).await
})
}
+
+#[rustler::nif]
+fn admin_get_table_info<'a>(
+ env: Env<'a>,
+ admin: ResourceArc<AdminResource>,
+ database_name: String,
+ table_name: String,
+) -> Term<'a> {
+ async_nif::spawn_task_with_result(env, async move {
+ let table_path = TablePath::new(database_name, table_name);
+ let table_info = admin.inner.get_table_info(&table_path).await?;
+ NifTableInfo::from_core(&table_info)
+ })
+}
+
+#[derive(NifStruct)]
+#[module = "Fluss.SchemaInfo"]
+pub struct NifSchemaInfo {
+ pub schema: NifSchema,
+ pub schema_id: i32,
+}
+
+impl NifSchemaInfo {
+ pub fn from_core(info: &SchemaInfo) -> Result<Self, Error> {
+ Ok(Self {
+ schema: NifSchema::from_core(info.schema())?,
+ schema_id: info.schema_id(),
+ })
+ }
+}
+
+#[rustler::nif]
+fn admin_get_table_schema<'a>(
+ env: Env<'a>,
+ admin: ResourceArc<AdminResource>,
+ database_name: String,
+ table_name: String,
+ schema_id: Option<i32>,
+) -> Term<'a> {
+ async_nif::spawn_task_with_result(env, async move {
+ let table_path = TablePath::new(database_name, table_name);
+ let schema_info = admin.inner.get_table_schema(&table_path,
schema_id).await?;
+ NifSchemaInfo::from_core(&schema_info)
+ })
+}
diff --git a/bindings/elixir/native/fluss_nif/src/schema.rs
b/bindings/elixir/native/fluss_nif/src/schema.rs
index 5d61d29d..410334fc 100644
--- a/bindings/elixir/native/fluss_nif/src/schema.rs
+++ b/bindings/elixir/native/fluss_nif/src/schema.rs
@@ -16,6 +16,7 @@
// under the License.
use crate::atoms::to_nif_err;
+use fluss::error::Error;
use fluss::metadata::{self, DataTypes, Schema, TableDescriptor};
use rustler::{NifStruct, NifTaggedEnum, ResourceArc};
@@ -73,6 +74,32 @@ fn to_fluss_type(dt: &DataType) -> metadata::DataType {
}
}
+fn from_fluss_type(dt: &metadata::DataType) -> Result<DataType, Error> {
+ match dt {
+ metadata::DataType::Boolean(_) => Ok(DataType::Boolean),
+ metadata::DataType::TinyInt(_) => Ok(DataType::Tinyint),
+ metadata::DataType::SmallInt(_) => Ok(DataType::Smallint),
+ metadata::DataType::Int(_) => Ok(DataType::Int),
+ metadata::DataType::BigInt(_) => Ok(DataType::Bigint),
+ metadata::DataType::Float(_) => Ok(DataType::Float),
+ metadata::DataType::Double(_) => Ok(DataType::Double),
+ metadata::DataType::String(_) => Ok(DataType::String),
+ metadata::DataType::Bytes(_) => Ok(DataType::Bytes),
+ metadata::DataType::Date(_) => Ok(DataType::Date),
+ metadata::DataType::Time(_) => Ok(DataType::Time),
+ metadata::DataType::Timestamp(_) => Ok(DataType::Timestamp),
+ metadata::DataType::TimestampLTz(_) => Ok(DataType::TimestampLtz),
+ metadata::DataType::Decimal(d) => Ok(DataType::Decimal(d.precision(),
d.scale())),
+ metadata::DataType::Char(c) => Ok(DataType::Char(c.length())),
+ metadata::DataType::Binary(b) => Ok(DataType::Binary(b.length())),
+ metadata::DataType::Array(_) | metadata::DataType::Map(_) |
metadata::DataType::Row(_) => {
+ Err(Error::UnsupportedOperation {
+ message: format!("data type {dt:?} is not supported by the
Elixir bindings"),
+ })
+ }
+ }
+}
+
/// Decoded from `%Fluss.Schema{}` Elixir struct.
#[derive(NifStruct)]
#[module = "Fluss.Schema"]
@@ -81,6 +108,27 @@ pub struct NifSchema {
pub primary_key: Vec<String>,
}
+impl NifSchema {
+ pub fn from_core(schema: &Schema) -> Result<NifSchema, Error> {
+ let mut columns: Vec<(String, DataType)> = Vec::new();
+
+ for col in schema.columns() {
+ columns.push((col.name().to_string(),
from_fluss_type(col.data_type())?));
+ }
+
+ let primary_key: Vec<String> = schema
+ .primary_key_column_names()
+ .iter()
+ .map(|s| s.to_string())
+ .collect();
+
+ Ok(Self {
+ columns,
+ primary_key,
+ })
+ }
+}
+
#[rustler::nif]
fn table_descriptor_new(
schema: NifSchema,
diff --git a/bindings/elixir/test/integration/admin_test.exs
b/bindings/elixir/test/integration/admin_test.exs
index e6a3ff73..8c34187a 100644
--- a/bindings/elixir/test/integration/admin_test.exs
+++ b/bindings/elixir/test/integration/admin_test.exs
@@ -235,4 +235,169 @@ defmodule Fluss.Integration.AdminTest do
refute Fluss.Admin.table_exists!(admin, @database, table)
end
end
+
+ describe "get_table_info/3" do
+ test "returns a %Fluss.TableInfo{} with the table's metadata", %{admin:
admin} do
+ table = "fluss_table_#{:rand.uniform(100_000)}"
+
+ schema =
+ Fluss.Schema.new()
+ |> Fluss.Schema.column("id", :int)
+ |> Fluss.Schema.column("name", :string)
+ |> Fluss.Schema.primary_key(["id"])
+
+ descriptor = Fluss.TableDescriptor.new!(schema, bucket_count: 4)
+ :ok = Fluss.Admin.create_table(admin, @database, table, descriptor, true)
+ on_exit(fn -> Fluss.Admin.drop_table(admin, @database, table, true) end)
+
+ assert {:ok, info} = Fluss.Admin.get_table_info(admin, @database, table)
+ assert %Fluss.TableInfo{} = info
+
+ # Fields we set explicitly — exact assertions.
+ assert info.database_name == @database
+ assert info.table_name == table
+ assert info.num_buckets == 4
+ assert info.has_primary_key == true
+ assert info.primary_keys == ["id"]
+
+ # Schema round-trips: columns in definition order, PK preserved.
+ assert info.schema == %Fluss.Schema{
+ columns: [{"id", :int}, {"name", :string}],
+ primary_key: ["id"]
+ }
+
+ assert info.is_partitioned == false
+ assert info.is_auto_partitioned == false
+ assert info.partition_keys == []
+ assert info.comment == nil
+
+ assert info.physical_primary_keys == ["id"]
+ assert info.bucket_keys == ["id"]
+ assert info.has_bucket_key == true
+ assert info.is_default_bucket_key == true
+
+ # Server-assigned — sanity only, values aren't predictable.
+ assert is_integer(info.table_id) and info.table_id > 0
+ assert is_integer(info.schema_id) and info.schema_id >= 1
+ assert is_integer(info.created_time) and info.created_time > 0
+ assert is_integer(info.modified_time) and info.modified_time > 0
+
+ assert is_map(info.properties)
+ assert is_map(info.custom_properties)
+ end
+
+ test "exposes a composite primary key", %{admin: admin} do
+ table = "fluss_table_#{:rand.uniform(100_000)}"
+
+ schema =
+ Fluss.Schema.new()
+ |> Fluss.Schema.column("id", :int)
+ |> Fluss.Schema.column("region", :string)
+ |> Fluss.Schema.column("name", :string)
+ |> Fluss.Schema.primary_key(["id", "region"])
+
+ descriptor = Fluss.TableDescriptor.new!(schema, bucket_count: 4)
+ :ok = Fluss.Admin.create_table(admin, @database, table, descriptor, true)
+ on_exit(fn -> Fluss.Admin.drop_table(admin, @database, table, true) end)
+
+ assert {:ok, info} = Fluss.Admin.get_table_info(admin, @database, table)
+
+ assert info.has_primary_key == true
+ assert info.primary_keys == ["id", "region"]
+ assert info.physical_primary_keys == ["id", "region"]
+
+ assert info.schema == %Fluss.Schema{
+ columns: [{"id", :int}, {"region", :string}, {"name", :string}],
+ primary_key: ["id", "region"]
+ }
+ end
+
+ test "returns {:error, %Fluss.Error{}} for a non-existent table", %{admin:
admin} do
+ table = "fluss_table_#{:rand.uniform(100_000)}"
+ assert {:error, %Fluss.Error{}} = Fluss.Admin.get_table_info(admin,
@database, table)
+ end
+ end
+
+ describe "get_table_info!/3" do
+ test "returns the %Fluss.TableInfo{} directly without an :ok tuple",
%{admin: admin} do
+ table = "fluss_table_#{:rand.uniform(100_000)}"
+
+ schema =
+ Fluss.Schema.new()
+ |> Fluss.Schema.column("id", :int)
+ |> Fluss.Schema.column("name", :string)
+
+ descriptor = Fluss.TableDescriptor.new!(schema)
+ :ok = Fluss.Admin.create_table(admin, @database, table, descriptor, true)
+ on_exit(fn -> Fluss.Admin.drop_table(admin, @database, table, true) end)
+
+ info = Fluss.Admin.get_table_info!(admin, @database, table)
+ assert %Fluss.TableInfo{} = info
+ assert info.table_name == table
+ end
+
+ test "raises Fluss.Error for a non-existent table", %{admin: admin} do
+ table = "fluss_table_#{:rand.uniform(100_000)}"
+
+ assert_raise Fluss.Error, fn ->
+ Fluss.Admin.get_table_info!(admin, @database, table)
+ end
+ end
+ end
+
+ describe "get_table_schema/4" do
+ test "returns the current schema as %Fluss.SchemaInfo{}", %{admin: admin}
do
+ table = "fluss_table_#{:rand.uniform(100_000)}"
+
+ schema =
+ Fluss.Schema.new()
+ |> Fluss.Schema.column("id", :int)
+ |> Fluss.Schema.column("name", :string)
+ |> Fluss.Schema.primary_key(["id"])
+
+ descriptor = Fluss.TableDescriptor.new!(schema)
+ :ok = Fluss.Admin.create_table(admin, @database, table, descriptor, true)
+ on_exit(fn -> Fluss.Admin.drop_table(admin, @database, table, true) end)
+
+ assert {:ok, %Fluss.SchemaInfo{} = info} =
+ Fluss.Admin.get_table_schema(admin, @database, table)
+
+ assert info.schema == %Fluss.Schema{
+ columns: [{"id", :int}, {"name", :string}],
+ primary_key: ["id"]
+ }
+
+ assert is_integer(info.schema_id) and info.schema_id >= 1
+ end
+
+ test "fetches a specific schema version by id", %{admin: admin} do
+ table = "fluss_table_#{:rand.uniform(100_000)}"
+
+ schema =
+ Fluss.Schema.new()
+ |> Fluss.Schema.column("id", :int)
+ |> Fluss.Schema.column("name", :string)
+ |> Fluss.Schema.primary_key(["id"])
+
+ descriptor = Fluss.TableDescriptor.new!(schema)
+ :ok = Fluss.Admin.create_table(admin, @database, table, descriptor, true)
+ on_exit(fn -> Fluss.Admin.drop_table(admin, @database, table, true) end)
+
+ # Without alter_table there's only one version, so requesting its id
+ # explicitly must return the same SchemaInfo as the latest (nil) lookup.
+ {:ok, current} = Fluss.Admin.get_table_schema(admin, @database, table)
+
+ assert {:ok, by_id} =
+ Fluss.Admin.get_table_schema(admin, @database, table,
current.schema_id)
+
+ assert by_id == current
+ end
+
+ test "returns {:error, %Fluss.Error{}} for a non-existent table", %{admin:
admin} do
+ table = "fluss_table_#{:rand.uniform(100_000)}"
+
+ assert {:error, %Fluss.Error{}} =
+ Fluss.Admin.get_table_schema(admin, @database, table)
+ end
+ end
end
