Elixir bindings initial PR

.licenserc.yaml

@@ -27,6 +27,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/fluss-test-cluster", "crates/examples", "bindings/python", "bindings/cpp"]
+members = ["crates/fluss", "crates/fluss-test-cluster", "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,52 @@
+# 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.new()
+        |> Fluss.Schema.column("ts", :bigint)
+        |> Fluss.Schema.column("message", :string)
+
+      :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,111 @@
+# 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")
+
+      schema = Fluss.Schema.new() |> Fluss.Schema.column("ts", :bigint)
+      descriptor = Fluss.TableDescriptor.new!(schema)
+      :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 new(conn) do
+      {:ok, admin} -> admin
+      {:error, reason} -> raise "failed to create admin: #{reason}"
+    end
+  end
+
+  @spec create_database(t(), String.t(), boolean()) :: :ok | {:error, String.t()}
+  def create_database(admin, name, ignore_if_exists \\ true) do
+    admin
+    |> Native.admin_create_database(name, ignore_if_exists)
+    |> Native.await_nif()
+  end
+
+  @spec drop_database(t(), String.t(), boolean()) :: :ok | {:error, String.t()}
+  def drop_database(admin, name, ignore_if_not_exists \\ true) do
+    admin
+    |> Native.admin_drop_database(name, ignore_if_not_exists)
+    |> Native.await_nif()
+  end
+
+  @spec list_databases(t()) :: {:ok, [String.t()]} | {:error, String.t()}
+  def list_databases(admin) do
+    admin
+    |> Native.admin_list_databases()
+    |> Native.await_nif()
+  end
+
+  @spec list_databases!(t()) :: [String.t()]
+  def list_databases!(admin) do
+    case list_databases(admin) do
+      {:ok, dbs} -> dbs
+      {:error, reason} -> raise "failed to list databases: #{reason}"
+    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
+    admin
+    |> Native.admin_create_table(database, table, descriptor, ignore_if_exists)
+    |> Native.await_nif()
+  end
+
+  @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
+    admin
+    |> Native.admin_drop_table(database, table, ignore_if_not_exists)
+    |> Native.await_nif()
+  end
+
+  @spec list_tables(t(), String.t()) :: {:ok, [String.t()]} | {:error, String.t()}
+  def list_tables(admin, database) do
+    admin
+    |> Native.admin_list_tables(database)
+    |> Native.await_nif()
+  end
+
+  @spec list_tables!(t(), String.t()) :: [String.t()]
+  def list_tables!(admin, database) do
+    case list_tables(admin, database) do
+      {:ok, tables} -> tables
+      {:error, reason} -> raise "failed to list tables: #{reason}"
+    end
+  end
+end