Elixir bindings initial PR

.licenserc.yaml

@@ -26,6 +26,7 @@ header:
     - 'NOTICE'
     - 'DISCLAIMER'
     - 'bindings/python/fluss/py.typed'
+    - '**/mix.lock'
     - 'website/**'
     - '**/*.md'
     - '**/DEPENDENCIES.*.tsv'

Cargo.toml

@@ -28,7 +28,7 @@ keywords = ["fluss", "streaming-storage", "datalake"]
 
 [workspace]
 resolver = "2"
-members = ["crates/fluss", "crates/examples", "bindings/python", "bindings/cpp"]
+members = ["crates/fluss", "crates/examples", "bindings/python", "bindings/cpp", "bindings/elixir/native/fluss_nif"]
 
 [workspace.dependencies]
 fluss = { package = "fluss-rs", version = "0.2.0", path = "crates/fluss", features = ["storage-all"] }

bindings/elixir/.formatter.exs

@@ -0,0 +1,20 @@
+# 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.
+
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

bindings/elixir/.gitignore

@@ -0,0 +1,9 @@
+# Elixir build artifacts
+_build/
+deps/
+
+# Generated NIF shared library
+priv/native/
+
+# Crash dumps
+erl_crash.dump

bindings/elixir/README.md

@@ -0,0 +1,60 @@
+# Fluss Elixir Client
+
+Elixir client for [Apache Fluss (Incubating)](https://fluss.apache.org/), built on the official Rust client via [Rustler](https://github.com/rusterlium/rustler) NIFs.
+
+Currently supports **log tables** (append + scan). Primary key (KV) table support is planned.
+
+## Requirements
+
+- Elixir >= 1.15
+- Rust stable toolchain (for compiling the NIF)
+
+## Quick Start
+
+```elixir
+config = Fluss.Config.new("localhost:9123")
+conn = Fluss.Connection.new!(config)
+admin = Fluss.Admin.new!(conn)
+
+schema =
+  Fluss.Schema.build()
+  |> Fluss.Schema.column("ts", :bigint)
+  |> Fluss.Schema.column("message", :string)
+  |> Fluss.Schema.build!()
+
+:ok = Fluss.Admin.create_table(admin, "my_db", "events", Fluss.TableDescriptor.new!(schema))
+
+table = Fluss.Table.get!(conn, "my_db", "events")
+writer = Fluss.AppendWriter.new!(table)
+Fluss.AppendWriter.append(writer, [1_700_000_000, "hello"])
+:ok = Fluss.AppendWriter.flush(writer)
+
+scanner = Fluss.LogScanner.new!(table)
+:ok = Fluss.LogScanner.subscribe(scanner, 0, Fluss.earliest_offset())
+:ok = Fluss.LogScanner.poll(scanner, 5_000)
+
+receive do
+  {:fluss_records, records} ->
+    for record <- records, do: IO.inspect(record[:row])
+end
+```
+
+## Data Types
+
+Simple: `:boolean`, `:tinyint`, `:smallint`, `:int`, `:bigint`, `:float`, `:double`, `:string`, `:bytes`, `:date`, `:time`, `:timestamp`, `:timestamp_ltz`
+
+Parameterized: `{:decimal, precision, scale}`, `{:char, length}`, `{:binary, length}`
+
+## Development
+
+```bash
+cd bindings/elixir
+mix test                        # unit tests
+mix test --include integration  # starts Docker cluster
+```
+
+Set `FLUSS_BOOTSTRAP_SERVERS` to use an existing cluster.
+
+## License
+
+Apache License 2.0

bindings/elixir/lib/fluss.ex

@@ -0,0 +1,53 @@
+# 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 do
+  @moduledoc """
+  Elixir client for Apache Fluss (Incubating).
+
+  ## Examples
+
+      config = Fluss.Config.new("localhost:9123")
+      conn = Fluss.Connection.new!(config)
+      admin = Fluss.Admin.new!(conn)
+
+      schema =
+        Fluss.Schema.build()
+        |> Fluss.Schema.column("ts", :bigint)
+        |> Fluss.Schema.column("message", :string)
+        |> Fluss.Schema.build!()
+
+      :ok = Fluss.Admin.create_table(admin, "my_db", "events", Fluss.TableDescriptor.new!(schema))
+
+      table = Fluss.Table.get!(conn, "my_db", "events")
+      writer = Fluss.AppendWriter.new!(table)
+      Fluss.AppendWriter.append(writer, [1_700_000_000, "hello"])
+      :ok = Fluss.AppendWriter.flush(writer)
+
+      scanner = Fluss.LogScanner.new!(table)
+      :ok = Fluss.LogScanner.subscribe(scanner, 0, Fluss.earliest_offset())
+      :ok = Fluss.LogScanner.poll(scanner, 5_000)
+      receive do
+        {:fluss_records, records} -> records
+      end
+
+  """
+
+  alias Fluss.Native
+
+  def earliest_offset, do: Native.earliest_offset()
+end

bindings/elixir/lib/fluss/admin.ex

@@ -0,0 +1,98 @@
+# 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.Admin do
+  @moduledoc """
+  Admin client for DDL operations (create/drop databases and tables).
+
+  ## Examples
+
+      admin = Fluss.Admin.new!(conn)
+      :ok = Fluss.Admin.create_database(admin, "my_db")
+      :ok = Fluss.Admin.create_table(admin, "my_db", "events", descriptor)
+
+  """
+
+  alias Fluss.Native
+
+  @type t :: reference()
+
+  @spec new(Fluss.Connection.t()) :: {:ok, t()} | {:error, String.t()}
+  def new(conn) do
+    case Native.admin_new(conn) do
+      {:error, _} = err -> err
+      admin -> {:ok, admin}
+    end
+  end
+
+  @spec new!(Fluss.Connection.t()) :: t()
+  def new!(conn) do
+    case Native.admin_new(conn) do
+      {:error, reason} -> raise "failed to create admin: #{reason}"
+      admin -> admin
+    end
+  end
+
+  @spec create_database(t(), String.t(), boolean()) :: :ok | {:error, String.t()}
+  def create_database(admin, name, ignore_if_exists \\ true),
+    do: Native.admin_create_database(admin, name, ignore_if_exists)
+
+  @spec drop_database(t(), String.t(), boolean()) :: :ok | {:error, String.t()}
+  def drop_database(admin, name, ignore_if_not_exists \\ true),
+    do: Native.admin_drop_database(admin, name, ignore_if_not_exists)
+
+  @spec list_databases(t()) :: {:ok, [String.t()]} | {:error, String.t()}
+  def list_databases(admin) do
+    case Native.admin_list_databases(admin) do
+      {:error, _} = err -> err
+      dbs -> {:ok, dbs}
+    end
+  end
+
+  @spec list_databases!(t()) :: [String.t()]
+  def list_databases!(admin) do
+    case Native.admin_list_databases(admin) do
+      {:error, reason} -> raise "failed to list databases: #{reason}"
+      dbs -> dbs
+    end
+  end
+
+  @spec create_table(t(), String.t(), String.t(), Fluss.TableDescriptor.t(), boolean()) ::
+          :ok | {:error, String.t()}
+  def create_table(admin, database, table, descriptor, ignore_if_exists \\ true),
+    do: Native.admin_create_table(admin, database, table, descriptor, ignore_if_exists)
+
+  @spec drop_table(t(), String.t(), String.t(), boolean()) :: :ok | {:error, String.t()}
+  def drop_table(admin, database, table, ignore_if_not_exists \\ true),
+    do: Native.admin_drop_table(admin, database, table, ignore_if_not_exists)
+
+  @spec list_tables(t(), String.t()) :: {:ok, [String.t()]} | {:error, String.t()}
+  def list_tables(admin, database) do
+    case Native.admin_list_tables(admin, database) do
+      {:error, _} = err -> err
+      tables -> {:ok, tables}
+    end
+  end
+
+  @spec list_tables!(t(), String.t()) :: [String.t()]
+  def list_tables!(admin, database) do
+    case Native.admin_list_tables(admin, database) do
+      {:error, reason} -> raise "failed to list tables: #{reason}"
+      tables -> tables
+    end
+  end
+end

bindings/elixir/lib/fluss/append_writer.ex

@@ -0,0 +1,71 @@
+# 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.AppendWriter do
+  @moduledoc """
+  Writer for appending records to a log table.
+
+  Values are passed as a list in column order. Use `nil` for null values.
+  `append/2` returns a `Fluss.WriteHandle` — drop it for fire-and-forget,
+  or call `Fluss.WriteHandle.wait/1` for per-record acknowledgment.
+
+  ## Examples
+
+      writer = Fluss.AppendWriter.new!(table)
+
+      # Fire-and-forget
+      Fluss.AppendWriter.append(writer, [1_700_000_000, "hello"])
+      Fluss.AppendWriter.append(writer, [1_700_000_001, "world"])
+      :ok = Fluss.AppendWriter.flush(writer)
+
+      # Per-record ack
+      {:ok, handle} = Fluss.AppendWriter.append(writer, [1_700_000_002, "critical"])
+      :ok = Fluss.WriteHandle.wait(handle)
+
+  """
+
+  alias Fluss.Native
+
+  @type t :: reference()
+
+  @spec new(Fluss.Table.t()) :: {:ok, t()} | {:error, String.t()}
+  def new(table) do
+    case Native.append_writer_new(table) do
+      {:error, _} = err -> err
+      w -> {:ok, w}
+    end
+  end
+
+  @spec new!(Fluss.Table.t()) :: t()
+  def new!(table) do
+    case Native.append_writer_new(table) do
+      {:error, reason} -> raise "failed to create append writer: #{reason}"
+      w -> w
+    end
+  end
+
+  @spec append(t(), list()) :: {:ok, Fluss.WriteHandle.t()} | {:error, String.t()}
+  def append(writer, values) when is_list(values) do
+    case Native.append_writer_append(writer, values) do
+      {:error, _} = err -> err
+      handle -> {:ok, handle}
+    end
+  end
+
+  @spec flush(t()) :: :ok | {:error, String.t()}
+  def flush(writer), do: Native.append_writer_flush(writer)
+end

bindings/elixir/lib/fluss/config.ex

@@ -0,0 +1,58 @@
+# 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.Config do
+  @moduledoc """
+  Client configuration for connecting to a Fluss cluster.
+
+  ## Examples
+
+      config = Fluss.Config.new("localhost:9123")
+
+      config =
+        Fluss.Config.default()
+        |> Fluss.Config.set_bootstrap_servers("host1:9123,host2:9123")
+        |> Fluss.Config.set_writer_batch_size(1_048_576)
+
+  """
+
+  alias Fluss.Native
+
+  @type t :: reference()
+
+  @spec new(String.t()) :: t()
+  def new(bootstrap_servers) when is_binary(bootstrap_servers) do
+    Native.config_new(bootstrap_servers)
+  end
+
+  @spec default() :: t()
+  def default, do: Native.config_default()
+
+  @spec set_bootstrap_servers(t(), String.t()) :: t()
+  def set_bootstrap_servers(config, servers),
+    do: Native.config_set_bootstrap_servers(config, servers)
+
+  @spec set_writer_batch_size(t(), integer()) :: t()
+  def set_writer_batch_size(config, size), do: Native.config_set_writer_batch_size(config, size)
+
+  @spec set_writer_batch_timeout_ms(t(), integer()) :: t()
+  def set_writer_batch_timeout_ms(config, ms),
+    do: Native.config_set_writer_batch_timeout_ms(config, ms)
+
+  @spec get_bootstrap_servers(t()) :: String.t()
+  def get_bootstrap_servers(config), do: Native.config_get_bootstrap_servers(config)
+end