diff --git a/.mvn/wrapper/maven-wrapper.properties b/.mvn/wrapper/maven-wrapper.properties
new file mode 100644
index 00000000..2f94e616
--- /dev/null
+++ b/.mvn/wrapper/maven-wrapper.properties
@@ -0,0 +1,19 @@
+# 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.
+wrapperVersion=3.3.2
+distributionType=only-script
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.10/apache-maven-3.9.10-bin.zip
diff --git a/BUILD.md b/BUILD.md
index 91c6e4e6..cd251e75 100644
--- a/BUILD.md
+++ b/BUILD.md
@@ -10,11 +10,11 @@
 You can install the SDK artifacts in your local maven cache by running the following command in project folder.
 During the build process, the working directory will be cleared, tests will be run, artifacts will be built and copied to the local repository.
 ```
-mvn clean install
+./mvnw clean install
 ```
 
 If you don't need the test executions, just disable them
 ```
-mvn clean install -DskipTests=true
+./mvnw clean install -DskipTests=true
 ```
 
diff --git a/mvnw b/mvnw
new file mode 100755
index 00000000..19529ddf
--- /dev/null
+++ b/mvnw
@@ -0,0 +1,259 @@
+#!/bin/sh
+# ----------------------------------------------------------------------------
+# 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.
+# ----------------------------------------------------------------------------
+
+# ----------------------------------------------------------------------------
+# Apache Maven Wrapper startup batch script, version 3.3.2
+#
+# Optional ENV vars
+# -----------------
+#   JAVA_HOME - location of a JDK home dir, required when download maven via java source
+#   MVNW_REPOURL - repo url base for downloading maven distribution
+#   MVNW_USERNAME/MVNW_PASSWORD - user and password for downloading maven
+#   MVNW_VERBOSE - true: enable verbose log; debug: trace the mvnw script; others: silence the output
+# ----------------------------------------------------------------------------
+
+set -euf
+[ "${MVNW_VERBOSE-}" != debug ] || set -x
+
+# OS specific support.
+native_path() { printf %s\\n "$1"; }
+case "$(uname)" in
+CYGWIN* | MINGW*)
+  [ -z "${JAVA_HOME-}" ] || JAVA_HOME="$(cygpath --unix "$JAVA_HOME")"
+  native_path() { cygpath --path --windows "$1"; }
+  ;;
+esac
+
+# set JAVACMD and JAVACCMD
+set_java_home() {
+  # For Cygwin and MinGW, ensure paths are in Unix format before anything is touched
+  if [ -n "${JAVA_HOME-}" ]; then
+    if [ -x "$JAVA_HOME/jre/sh/java" ]; then
+      # IBM's JDK on AIX uses strange locations for the executables
+      JAVACMD="$JAVA_HOME/jre/sh/java"
+      JAVACCMD="$JAVA_HOME/jre/sh/javac"
+    else
+      JAVACMD="$JAVA_HOME/bin/java"
+      JAVACCMD="$JAVA_HOME/bin/javac"
+
+      if [ ! -x "$JAVACMD" ] || [ ! -x "$JAVACCMD" ]; then
+        echo "The JAVA_HOME environment variable is not defined correctly, so mvnw cannot run." >&2
+        echo "JAVA_HOME is set to \"$JAVA_HOME\", but \"\$JAVA_HOME/bin/java\" or \"\$JAVA_HOME/bin/javac\" does not exist." >&2
+        return 1
+      fi
+    fi
+  else
+    JAVACMD="$(
+      'set' +e
+      'unset' -f command 2>/dev/null
+      'command' -v java
+    )" || :
+    JAVACCMD="$(
+      'set' +e
+      'unset' -f command 2>/dev/null
+      'command' -v javac
+    )" || :
+
+    if [ ! -x "${JAVACMD-}" ] || [ ! -x "${JAVACCMD-}" ]; then
+      echo "The java/javac command does not exist in PATH nor is JAVA_HOME set, so mvnw cannot run." >&2
+      return 1
+    fi
+  fi
+}
+
+# hash string like Java String::hashCode
+hash_string() {
+  str="${1:-}" h=0
+  while [ -n "$str" ]; do
+    char="${str%"${str#?}"}"
+    h=$(((h * 31 + $(LC_CTYPE=C printf %d "'$char")) % 4294967296))
+    str="${str#?}"
+  done
+  printf %x\\n $h
+}
+
+verbose() { :; }
+[ "${MVNW_VERBOSE-}" != true ] || verbose() { printf %s\\n "${1-}"; }
+
+die() {
+  printf %s\\n "$1" >&2
+  exit 1
+}
+
+trim() {
+  # MWRAPPER-139:
+  #   Trims trailing and leading whitespace, carriage returns, tabs, and linefeeds.
+  #   Needed for removing poorly interpreted newline sequences when running in more
+  #   exotic environments such as mingw bash on Windows.
+  printf "%s" "${1}" | tr -d '[:space:]'
+}
+
+# parse distributionUrl and optional distributionSha256Sum, requires .mvn/wrapper/maven-wrapper.properties
+while IFS="=" read -r key value; do
+  case "${key-}" in
+  distributionUrl) distributionUrl=$(trim "${value-}") ;;
+  distributionSha256Sum) distributionSha256Sum=$(trim "${value-}") ;;
+  esac
+done <"${0%/*}/.mvn/wrapper/maven-wrapper.properties"
+[ -n "${distributionUrl-}" ] || die "cannot read distributionUrl property in ${0%/*}/.mvn/wrapper/maven-wrapper.properties"
+
+case "${distributionUrl##*/}" in
+maven-mvnd-*bin.*)
+  MVN_CMD=mvnd.sh _MVNW_REPO_PATTERN=/maven/mvnd/
+  case "${PROCESSOR_ARCHITECTURE-}${PROCESSOR_ARCHITEW6432-}:$(uname -a)" in
+  *AMD64:CYGWIN* | *AMD64:MINGW*) distributionPlatform=windows-amd64 ;;
+  :Darwin*x86_64) distributionPlatform=darwin-amd64 ;;
+  :Darwin*arm64) distributionPlatform=darwin-aarch64 ;;
+  :Linux*x86_64*) distributionPlatform=linux-amd64 ;;
+  *)
+    echo "Cannot detect native platform for mvnd on $(uname)-$(uname -m), use pure java version" >&2
+    distributionPlatform=linux-amd64
+    ;;
+  esac
+  distributionUrl="${distributionUrl%-bin.*}-$distributionPlatform.zip"
+  ;;
+maven-mvnd-*) MVN_CMD=mvnd.sh _MVNW_REPO_PATTERN=/maven/mvnd/ ;;
+*) MVN_CMD="mvn${0##*/mvnw}" _MVNW_REPO_PATTERN=/org/apache/maven/ ;;
+esac
+
+# apply MVNW_REPOURL and calculate MAVEN_HOME
+# maven home pattern: ~/.m2/wrapper/dists/{apache-maven-<version>,maven-mvnd-<version>-<platform>}/<hash>
+[ -z "${MVNW_REPOURL-}" ] || distributionUrl="$MVNW_REPOURL$_MVNW_REPO_PATTERN${distributionUrl#*"$_MVNW_REPO_PATTERN"}"
+distributionUrlName="${distributionUrl##*/}"
+distributionUrlNameMain="${distributionUrlName%.*}"
+distributionUrlNameMain="${distributionUrlNameMain%-bin}"
+MAVEN_USER_HOME="${MAVEN_USER_HOME:-${HOME}/.m2}"
+MAVEN_HOME="${MAVEN_USER_HOME}/wrapper/dists/${distributionUrlNameMain-}/$(hash_string "$distributionUrl")"
+
+exec_maven() {
+  unset MVNW_VERBOSE MVNW_USERNAME MVNW_PASSWORD MVNW_REPOURL || :
+  exec "$MAVEN_HOME/bin/$MVN_CMD" "$@" || die "cannot exec $MAVEN_HOME/bin/$MVN_CMD"
+}
+
+if [ -d "$MAVEN_HOME" ]; then
+  verbose "found existing MAVEN_HOME at $MAVEN_HOME"
+  exec_maven "$@"
+fi
+
+case "${distributionUrl-}" in
+*?-bin.zip | *?maven-mvnd-?*-?*.zip) ;;
+*) die "distributionUrl is not valid, must match *-bin.zip or maven-mvnd-*.zip, but found '${distributionUrl-}'" ;;
+esac
+
+# prepare tmp dir
+if TMP_DOWNLOAD_DIR="$(mktemp -d)" && [ -d "$TMP_DOWNLOAD_DIR" ]; then
+  clean() { rm -rf -- "$TMP_DOWNLOAD_DIR"; }
+  trap clean HUP INT TERM EXIT
+else
+  die "cannot create temp dir"
+fi
+
+mkdir -p -- "${MAVEN_HOME%/*}"
+
+# Download and Install Apache Maven
+verbose "Couldn't find MAVEN_HOME, downloading and installing it ..."
+verbose "Downloading from: $distributionUrl"
+verbose "Downloading to: $TMP_DOWNLOAD_DIR/$distributionUrlName"
+
+# select .zip or .tar.gz
+if ! command -v unzip >/dev/null; then
+  distributionUrl="${distributionUrl%.zip}.tar.gz"
+  distributionUrlName="${distributionUrl##*/}"
+fi
+
+# verbose opt
+__MVNW_QUIET_WGET=--quiet __MVNW_QUIET_CURL=--silent __MVNW_QUIET_UNZIP=-q __MVNW_QUIET_TAR=''
+[ "${MVNW_VERBOSE-}" != true ] || __MVNW_QUIET_WGET='' __MVNW_QUIET_CURL='' __MVNW_QUIET_UNZIP='' __MVNW_QUIET_TAR=v
+
+# normalize http auth
+case "${MVNW_PASSWORD:+has-password}" in
+'') MVNW_USERNAME='' MVNW_PASSWORD='' ;;
+has-password) [ -n "${MVNW_USERNAME-}" ] || MVNW_USERNAME='' MVNW_PASSWORD='' ;;
+esac
+
+if [ -z "${MVNW_USERNAME-}" ] && command -v wget >/dev/null; then
+  verbose "Found wget ... using wget"
+  wget ${__MVNW_QUIET_WGET:+"$__MVNW_QUIET_WGET"} "$distributionUrl" -O "$TMP_DOWNLOAD_DIR/$distributionUrlName" || die "wget: Failed to fetch $distributionUrl"
+elif [ -z "${MVNW_USERNAME-}" ] && command -v curl >/dev/null; then
+  verbose "Found curl ... using curl"
+  curl ${__MVNW_QUIET_CURL:+"$__MVNW_QUIET_CURL"} -f -L -o "$TMP_DOWNLOAD_DIR/$distributionUrlName" "$distributionUrl" || die "curl: Failed to fetch $distributionUrl"
+elif set_java_home; then
+  verbose "Falling back to use Java to download"
+  javaSource="$TMP_DOWNLOAD_DIR/Downloader.java"
+  targetZip="$TMP_DOWNLOAD_DIR/$distributionUrlName"
+  cat >"$javaSource" <<-END
+	public class Downloader extends java.net.Authenticator
+	{
+	  protected java.net.PasswordAuthentication getPasswordAuthentication()
+	  {
+	    return new java.net.PasswordAuthentication( System.getenv( "MVNW_USERNAME" ), System.getenv( "MVNW_PASSWORD" ).toCharArray() );
+	  }
+	  public static void main( String[] args ) throws Exception
+	  {
+	    setDefault( new Downloader() );
+	    java.nio.file.Files.copy( java.net.URI.create( args[0] ).toURL().openStream(), java.nio.file.Paths.get( args[1] ).toAbsolutePath().normalize() );
+	  }
+	}
+	END
+  # For Cygwin/MinGW, switch paths to Windows format before running javac and java
+  verbose " - Compiling Downloader.java ..."
+  "$(native_path "$JAVACCMD")" "$(native_path "$javaSource")" || die "Failed to compile Downloader.java"
+  verbose " - Running Downloader.java ..."
+  "$(native_path "$JAVACMD")" -cp "$(native_path "$TMP_DOWNLOAD_DIR")" Downloader "$distributionUrl" "$(native_path "$targetZip")"
+fi
+
+# If specified, validate the SHA-256 sum of the Maven distribution zip file
+if [ -n "${distributionSha256Sum-}" ]; then
+  distributionSha256Result=false
+  if [ "$MVN_CMD" = mvnd.sh ]; then
+    echo "Checksum validation is not supported for maven-mvnd." >&2
+    echo "Please disable validation by removing 'distributionSha256Sum' from your maven-wrapper.properties." >&2
+    exit 1
+  elif command -v sha256sum >/dev/null; then
+    if echo "$distributionSha256Sum  $TMP_DOWNLOAD_DIR/$distributionUrlName" | sha256sum -c >/dev/null 2>&1; then
+      distributionSha256Result=true
+    fi
+  elif command -v shasum >/dev/null; then
+    if echo "$distributionSha256Sum  $TMP_DOWNLOAD_DIR/$distributionUrlName" | shasum -a 256 -c >/dev/null 2>&1; then
+      distributionSha256Result=true
+    fi
+  else
+    echo "Checksum validation was requested but neither 'sha256sum' or 'shasum' are available." >&2
+    echo "Please install either command, or disable validation by removing 'distributionSha256Sum' from your maven-wrapper.properties." >&2
+    exit 1
+  fi
+  if [ $distributionSha256Result = false ]; then
+    echo "Error: Failed to validate Maven distribution SHA-256, your Maven distribution might be compromised." >&2
+    echo "If you updated your Maven version, you need to update the specified distributionSha256Sum property." >&2
+    exit 1
+  fi
+fi
+
+# unzip and move
+if command -v unzip >/dev/null; then
+  unzip ${__MVNW_QUIET_UNZIP:+"$__MVNW_QUIET_UNZIP"} "$TMP_DOWNLOAD_DIR/$distributionUrlName" -d "$TMP_DOWNLOAD_DIR" || die "failed to unzip"
+else
+  tar xzf${__MVNW_QUIET_TAR:+"$__MVNW_QUIET_TAR"} "$TMP_DOWNLOAD_DIR/$distributionUrlName" -C "$TMP_DOWNLOAD_DIR" || die "failed to untar"
+fi
+printf %s\\n "$distributionUrl" >"$TMP_DOWNLOAD_DIR/$distributionUrlNameMain/mvnw.url"
+mv -- "$TMP_DOWNLOAD_DIR/$distributionUrlNameMain" "$MAVEN_HOME" || [ -d "$MAVEN_HOME" ] || die "fail to move MAVEN_HOME"
+
+clean || :
+exec_maven "$@"
diff --git a/mvnw.cmd b/mvnw.cmd
new file mode 100644
index 00000000..249bdf38
--- /dev/null
+++ b/mvnw.cmd
@@ -0,0 +1,149 @@
+<# : batch portion
+@REM ----------------------------------------------------------------------------
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@REM "License"); you may not use this file except in compliance
+@REM with the License.  You may obtain a copy of the License at
+@REM
+@REM    http://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM ----------------------------------------------------------------------------
+
+@REM ----------------------------------------------------------------------------
+@REM Apache Maven Wrapper startup batch script, version 3.3.2
+@REM
+@REM Optional ENV vars
+@REM   MVNW_REPOURL - repo url base for downloading maven distribution
+@REM   MVNW_USERNAME/MVNW_PASSWORD - user and password for downloading maven
+@REM   MVNW_VERBOSE - true: enable verbose log; others: silence the output
+@REM ----------------------------------------------------------------------------
+
+@IF "%__MVNW_ARG0_NAME__%"=="" (SET __MVNW_ARG0_NAME__=%~nx0)
+@SET __MVNW_CMD__=
+@SET __MVNW_ERROR__=
+@SET __MVNW_PSMODULEP_SAVE=%PSModulePath%
+@SET PSModulePath=
+@FOR /F "usebackq tokens=1* delims==" %%A IN (`powershell -noprofile "& {$scriptDir='%~dp0'; $script='%__MVNW_ARG0_NAME__%'; icm -ScriptBlock ([Scriptblock]::Create((Get-Content -Raw '%~f0'))) -NoNewScope}"`) DO @(
+  IF "%%A"=="MVN_CMD" (set __MVNW_CMD__=%%B) ELSE IF "%%B"=="" (echo %%A) ELSE (echo %%A=%%B)
+)
+@SET PSModulePath=%__MVNW_PSMODULEP_SAVE%
+@SET __MVNW_PSMODULEP_SAVE=
+@SET __MVNW_ARG0_NAME__=
+@SET MVNW_USERNAME=
+@SET MVNW_PASSWORD=
+@IF NOT "%__MVNW_CMD__%"=="" (%__MVNW_CMD__% %*)
+@echo Cannot start maven from wrapper >&2 && exit /b 1
+@GOTO :EOF
+: end batch / begin powershell #>
+
+$ErrorActionPreference = "Stop"
+if ($env:MVNW_VERBOSE -eq "true") {
+  $VerbosePreference = "Continue"
+}
+
+# calculate distributionUrl, requires .mvn/wrapper/maven-wrapper.properties
+$distributionUrl = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionUrl
+if (!$distributionUrl) {
+  Write-Error "cannot read distributionUrl property in $scriptDir/.mvn/wrapper/maven-wrapper.properties"
+}
+
+switch -wildcard -casesensitive ( $($distributionUrl -replace '^.*/','') ) {
+  "maven-mvnd-*" {
+    $USE_MVND = $true
+    $distributionUrl = $distributionUrl -replace '-bin\.[^.]*$',"-windows-amd64.zip"
+    $MVN_CMD = "mvnd.cmd"
+    break
+  }
+  default {
+    $USE_MVND = $false
+    $MVN_CMD = $script -replace '^mvnw','mvn'
+    break
+  }
+}
+
+# apply MVNW_REPOURL and calculate MAVEN_HOME
+# maven home pattern: ~/.m2/wrapper/dists/{apache-maven-<version>,maven-mvnd-<version>-<platform>}/<hash>
+if ($env:MVNW_REPOURL) {
+  $MVNW_REPO_PATTERN = if ($USE_MVND) { "/org/apache/maven/" } else { "/maven/mvnd/" }
+  $distributionUrl = "$env:MVNW_REPOURL$MVNW_REPO_PATTERN$($distributionUrl -replace '^.*'+$MVNW_REPO_PATTERN,'')"
+}
+$distributionUrlName = $distributionUrl -replace '^.*/',''
+$distributionUrlNameMain = $distributionUrlName -replace '\.[^.]*$','' -replace '-bin$',''
+$MAVEN_HOME_PARENT = "$HOME/.m2/wrapper/dists/$distributionUrlNameMain"
+if ($env:MAVEN_USER_HOME) {
+  $MAVEN_HOME_PARENT = "$env:MAVEN_USER_HOME/wrapper/dists/$distributionUrlNameMain"
+}
+$MAVEN_HOME_NAME = ([System.Security.Cryptography.MD5]::Create().ComputeHash([byte[]][char[]]$distributionUrl) | ForEach-Object {$_.ToString("x2")}) -join ''
+$MAVEN_HOME = "$MAVEN_HOME_PARENT/$MAVEN_HOME_NAME"
+
+if (Test-Path -Path "$MAVEN_HOME" -PathType Container) {
+  Write-Verbose "found existing MAVEN_HOME at $MAVEN_HOME"
+  Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
+  exit $?
+}
+
+if (! $distributionUrlNameMain -or ($distributionUrlName -eq $distributionUrlNameMain)) {
+  Write-Error "distributionUrl is not valid, must end with *-bin.zip, but found $distributionUrl"
+}
+
+# prepare tmp dir
+$TMP_DOWNLOAD_DIR_HOLDER = New-TemporaryFile
+$TMP_DOWNLOAD_DIR = New-Item -Itemtype Directory -Path "$TMP_DOWNLOAD_DIR_HOLDER.dir"
+$TMP_DOWNLOAD_DIR_HOLDER.Delete() | Out-Null
+trap {
+  if ($TMP_DOWNLOAD_DIR.Exists) {
+    try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+    catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+  }
+}
+
+New-Item -Itemtype Directory -Path "$MAVEN_HOME_PARENT" -Force | Out-Null
+
+# Download and Install Apache Maven
+Write-Verbose "Couldn't find MAVEN_HOME, downloading and installing it ..."
+Write-Verbose "Downloading from: $distributionUrl"
+Write-Verbose "Downloading to: $TMP_DOWNLOAD_DIR/$distributionUrlName"
+
+$webclient = New-Object System.Net.WebClient
+if ($env:MVNW_USERNAME -and $env:MVNW_PASSWORD) {
+  $webclient.Credentials = New-Object System.Net.NetworkCredential($env:MVNW_USERNAME, $env:MVNW_PASSWORD)
+}
+[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
+$webclient.DownloadFile($distributionUrl, "$TMP_DOWNLOAD_DIR/$distributionUrlName") | Out-Null
+
+# If specified, validate the SHA-256 sum of the Maven distribution zip file
+$distributionSha256Sum = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionSha256Sum
+if ($distributionSha256Sum) {
+  if ($USE_MVND) {
+    Write-Error "Checksum validation is not supported for maven-mvnd. `nPlease disable validation by removing 'distributionSha256Sum' from your maven-wrapper.properties."
+  }
+  Import-Module $PSHOME\Modules\Microsoft.PowerShell.Utility -Function Get-FileHash
+  if ((Get-FileHash "$TMP_DOWNLOAD_DIR/$distributionUrlName" -Algorithm SHA256).Hash.ToLower() -ne $distributionSha256Sum) {
+    Write-Error "Error: Failed to validate Maven distribution SHA-256, your Maven distribution might be compromised. If you updated your Maven version, you need to update the specified distributionSha256Sum property."
+  }
+}
+
+# unzip and move
+Expand-Archive "$TMP_DOWNLOAD_DIR/$distributionUrlName" -DestinationPath "$TMP_DOWNLOAD_DIR" | Out-Null
+Rename-Item -Path "$TMP_DOWNLOAD_DIR/$distributionUrlNameMain" -NewName $MAVEN_HOME_NAME | Out-Null
+try {
+  Move-Item -Path "$TMP_DOWNLOAD_DIR/$MAVEN_HOME_NAME" -Destination $MAVEN_HOME_PARENT | Out-Null
+} catch {
+  if (! (Test-Path -Path "$MAVEN_HOME" -PathType Container)) {
+    Write-Error "fail to move MAVEN_HOME"
+  }
+} finally {
+  try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+  catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+}
+
+Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
diff --git a/table/src/main/java/tech/ydb/table/values/DecimalValue.java b/table/src/main/java/tech/ydb/table/values/DecimalValue.java
index bc68308f..72ef4d09 100644
--- a/table/src/main/java/tech/ydb/table/values/DecimalValue.java
+++ b/table/src/main/java/tech/ydb/table/values/DecimalValue.java
@@ -269,6 +269,36 @@ public ValueProtos.Value toPb() {
         return ProtoValue.fromDecimal(high, low);
     }
 
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (!optional.isPresent()) {
+                throw new NullPointerException("Cannot compare value " + this + " with NULL");
+            }
+            return compareTo(optional.get());
+        }
+
+        if (!(other instanceof DecimalValue)) {
+            throw new IllegalArgumentException("Cannot compare DecimalValue with " + other.getClass().getSimpleName());
+        }
+
+        DecimalValue decimal = (DecimalValue) other;
+
+        // Fast way to compare decimals with the same scale or with special values
+        boolean isSpecial = isNan() || isInf() || isNegativeInf();
+        boolean otherIsSpecial = decimal.isNan() || decimal.isInf() || decimal.isNegativeInf();
+        if (isSpecial || otherIsSpecial || (getType().getScale() == decimal.getType().getScale())) {
+            return high != decimal.high ? Long.compare(high, decimal.high) : Long.compare(low, decimal.low);
+        }
+
+        return toBigDecimal().compareTo(decimal.toBigDecimal());
+    }
+
     /**
      * Write long to a big-endian buffer.
      */
diff --git a/table/src/main/java/tech/ydb/table/values/DictValue.java b/table/src/main/java/tech/ydb/table/values/DictValue.java
index 59c9d28b..921b4a6e 100644
--- a/table/src/main/java/tech/ydb/table/values/DictValue.java
+++ b/table/src/main/java/tech/ydb/table/values/DictValue.java
@@ -4,6 +4,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.TreeSet;
 
 import javax.annotation.Nullable;
 
@@ -116,4 +117,46 @@ public ValueProtos.Value toPb() {
         }
         return builder.build();
     }
+
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (!optional.isPresent()) {
+                throw new NullPointerException("Cannot compare value " + this + " with NULL");
+            }
+            return compareTo(optional.get());
+        }
+
+        if (!type.equals(other.getType())) {
+            throw new IllegalArgumentException("Cannot compare value " + type + " with " + other.getType());
+        }
+
+        DictValue otherDict = (DictValue) other;
+
+        // Sort entries by keys
+        Set<Value<?>> keys = new TreeSet<>();
+        keys.addAll(items.keySet());
+        keys.addAll(otherDict.keySet());
+
+        for (Value<?> key: keys) {
+            if (!otherDict.items.containsKey(key)) {
+                return 1;
+            }
+            if (!items.containsKey(key)) {
+                return -1;
+            }
+
+            int valueComparison = items.get(key).compareTo(otherDict.items.get(key));
+            if (valueComparison != 0) {
+                return valueComparison;
+            }
+        }
+
+        return 0;
+    }
 }
diff --git a/table/src/main/java/tech/ydb/table/values/ListValue.java b/table/src/main/java/tech/ydb/table/values/ListValue.java
index 94699b60..4292a68a 100644
--- a/table/src/main/java/tech/ydb/table/values/ListValue.java
+++ b/table/src/main/java/tech/ydb/table/values/ListValue.java
@@ -113,4 +113,31 @@ public ValueProtos.Value toPb() {
         }
         return builder.build();
     }
+
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (!optional.isPresent()) {
+                throw new NullPointerException("Cannot compare value " + this + " with NULL");
+            }
+            return compareTo(optional.get());
+        }
+
+        ListValue list = (ListValue) other;
+
+        int minLength = Math.min(items.length, list.items.length);
+        for (int i = 0; i < minLength; i++) {
+            int itemComparison = items[i].compareTo(list.items[i]);
+            if (itemComparison != 0) {
+                return itemComparison;
+            }
+        }
+
+        return Integer.compare(items.length, list.items.length);
+    }
 }
diff --git a/table/src/main/java/tech/ydb/table/values/NullValue.java b/table/src/main/java/tech/ydb/table/values/NullValue.java
index 8d363796..b4cf6ca6 100644
--- a/table/src/main/java/tech/ydb/table/values/NullValue.java
+++ b/table/src/main/java/tech/ydb/table/values/NullValue.java
@@ -32,4 +32,26 @@ public NullType getType() {
     public ValueProtos.Value toPb() {
         return ProtoValue.nullValue();
     }
+
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (!optional.isPresent()) {
+                return 0;
+            }
+            return compareTo(optional.get());
+        }
+
+        if (other instanceof VoidValue || other instanceof NullValue) {
+            // All VoidValue and NullValue  are equal
+            return 0;
+        }
+
+        throw new IllegalArgumentException("Cannot compare value " + getType() + " with " + other.getType());
+    }
 }
diff --git a/table/src/main/java/tech/ydb/table/values/OptionalValue.java b/table/src/main/java/tech/ydb/table/values/OptionalValue.java
index 71c8d3f1..dfb57355 100644
--- a/table/src/main/java/tech/ydb/table/values/OptionalValue.java
+++ b/table/src/main/java/tech/ydb/table/values/OptionalValue.java
@@ -96,4 +96,28 @@ public ValueProtos.Value toPb() {
 
         return ProtoValue.optional();
     }
+
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (optional.value == null) {
+                if (value == null) {
+                    return 0;
+                }
+                throw new NullPointerException("Cannot compare value " + value + " with NULL");
+            }
+            return compareTo(optional.value);
+        }
+
+        if (value == null) {
+            throw new NullPointerException("Cannot compare NULL with value " + other);
+        }
+
+        return value.compareTo(other);
+    }
 }
diff --git a/table/src/main/java/tech/ydb/table/values/PrimitiveValue.java b/table/src/main/java/tech/ydb/table/values/PrimitiveValue.java
index 10338715..923d0e25 100644
--- a/table/src/main/java/tech/ydb/table/values/PrimitiveValue.java
+++ b/table/src/main/java/tech/ydb/table/values/PrimitiveValue.java
@@ -20,6 +20,7 @@
 import com.google.protobuf.UnsafeByteOperations;
 
 import tech.ydb.proto.ValueProtos;
+import tech.ydb.table.utils.LittleEndian;
 import tech.ydb.table.values.proto.ProtoValue;
 
 
@@ -423,6 +424,117 @@ private static void checkType(PrimitiveType expected, PrimitiveType actual) {
         }
     }
 
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (!optional.isPresent()) {
+                throw new NullPointerException("Cannot compare value " + this + " with NULL");
+            }
+            return compareTo(optional.get());
+        }
+
+        if (!getType().equals(other.getType())) {
+            throw new IllegalArgumentException("Cannot compare value " + getType() + " with " + other.getType());
+        }
+
+        PrimitiveValue otherValue = (PrimitiveValue) other;
+
+        // Compare based on the actual primitive type
+        switch (getType()) {
+            case Bool:
+                return Boolean.compare(getBool(), otherValue.getBool());
+            case Int8:
+                return Byte.compare(getInt8(), otherValue.getInt8());
+            case Uint8:
+                return Integer.compare(getUint8(), otherValue.getUint8());
+            case Int16:
+                return Short.compare(getInt16(), otherValue.getInt16());
+            case Uint16:
+                return Integer.compare(getUint16(), otherValue.getUint16());
+            case Int32:
+                return Integer.compare(getInt32(), otherValue.getInt32());
+            case Uint32:
+                return Long.compare(getUint32(), otherValue.getUint32());
+            case Int64:
+                return Long.compare(getInt64(), otherValue.getInt64());
+            case Uint64:
+                return Long.compareUnsigned(getUint64(), otherValue.getUint64());
+            case Float:
+                return Float.compare(getFloat(), otherValue.getFloat());
+            case Double:
+                return Double.compare(getDouble(), otherValue.getDouble());
+            case Bytes:
+                return compareArrays(getBytesUnsafe(), otherValue.getBytesUnsafe());
+            case Yson:
+                return compareArrays(getYsonUnsafe(), otherValue.getYsonUnsafe());
+            case Text:
+                return getText().compareTo(otherValue.getText());
+            case Json:
+                return getJson().compareTo(otherValue.getJson());
+            case JsonDocument:
+                return getJsonDocument().compareTo(otherValue.getJsonDocument());
+            case Uuid:
+                return compareUUID(this, otherValue);
+            case Date:
+                return getDate().compareTo(otherValue.getDate());
+            case Date32:
+                return getDate32().compareTo(otherValue.getDate32());
+            case Datetime:
+                return getDatetime().compareTo(otherValue.getDatetime());
+            case Datetime64:
+                return getDatetime64().compareTo(otherValue.getDatetime64());
+            case Timestamp:
+                return getTimestamp().compareTo(otherValue.getTimestamp());
+            case Timestamp64:
+                return getTimestamp64().compareTo(otherValue.getTimestamp64());
+            case Interval:
+                return getInterval().compareTo(otherValue.getInterval());
+            case Interval64:
+                return getInterval64().compareTo(otherValue.getInterval64());
+            case TzDate:
+                return getTzDate().compareTo(otherValue.getTzDate());
+            case TzDatetime:
+                return getTzDatetime().compareTo(otherValue.getTzDatetime());
+            case TzTimestamp:
+                return getTzTimestamp().compareTo(otherValue.getTzTimestamp());
+            default:
+                throw new UnsupportedOperationException("Comparison not supported for type: " + getType());
+        }
+    }
+
+    @SuppressWarnings("deprecation")
+    private static int compareUUID(PrimitiveValue a, PrimitiveValue b) {
+        long ah = LittleEndian.bswap(a.getUuidHigh());
+        long bh = LittleEndian.bswap(b.getUuidHigh());
+        long al = LittleEndian.bswap(a.getUuidLow());
+        long bl = LittleEndian.bswap(b.getUuidLow());
+
+        return (al != bl) ? Long.compareUnsigned(al, bl) : Long.compareUnsigned(ah, bh);
+    }
+
+    private static int compareArrays(byte[] a, byte[] b) {
+        if (a == b) {
+            return 0;
+        }
+
+        int i = 0;
+        int len = Math.min(a.length, b.length);
+        while (i < len && a[i] == b[i]) {
+            i++;
+        }
+
+        if (i < len) {
+            return Byte.compare(a[i], b[i]);
+        }
+
+        return a.length - b.length;
+    }
+
     // -- implementations --
 
     private static final class Bool extends PrimitiveValue {
diff --git a/table/src/main/java/tech/ydb/table/values/StructValue.java b/table/src/main/java/tech/ydb/table/values/StructValue.java
index 7c34d37a..2e24fe2f 100644
--- a/table/src/main/java/tech/ydb/table/values/StructValue.java
+++ b/table/src/main/java/tech/ydb/table/values/StructValue.java
@@ -134,6 +134,35 @@ public ValueProtos.Value toPb() {
         return builder.build();
     }
 
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (!optional.isPresent()) {
+                throw new NullPointerException("Cannot compare value " + this + " with NULL");
+            }
+            return compareTo(optional.get());
+        }
+
+        if (!type.equals(other.getType())) {
+            throw new IllegalArgumentException("Cannot compare value " + type + " with " + other.getType());
+        }
+
+        StructValue struct = (StructValue) other;
+        for (int i = 0; i < type.getMembersCount(); i++) {
+            int memberComparison = members[i].compareTo(struct.members[i]);
+            if (memberComparison != 0) {
+                return memberComparison;
+            }
+        }
+
+        return 0;
+    }
+
     private static StructValue newStruct(String[] names, Value<?>[] values) {
         Arrays2.sortBothByFirst(names, values);
         final Type[] types = new Type[values.length];
diff --git a/table/src/main/java/tech/ydb/table/values/TupleValue.java b/table/src/main/java/tech/ydb/table/values/TupleValue.java
index 9b80a1b2..d5b08fb0 100644
--- a/table/src/main/java/tech/ydb/table/values/TupleValue.java
+++ b/table/src/main/java/tech/ydb/table/values/TupleValue.java
@@ -145,4 +145,34 @@ private static TupleValue fromArray(Value<?>... items) {
         }
         return new TupleValue(TupleType.ofOwn(types), items);
     }
+
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (!optional.isPresent()) {
+                throw new NullPointerException("Cannot compare value " + this + " with NULL");
+            }
+            return compareTo(optional.get());
+        }
+
+        if (!type.equals(other.getType())) {
+            throw new IllegalArgumentException("Cannot compare value " + type + " with " + other.getType());
+        }
+
+        TupleValue otherTuple = (TupleValue) other;
+
+        for (int i = 0; i < getType().getElementsCount(); i++) {
+            int itemComparison = items[i].compareTo(otherTuple.items[i]);
+            if (itemComparison != 0) {
+                return itemComparison;
+            }
+        }
+
+        return 0;
+    }
 }
diff --git a/table/src/main/java/tech/ydb/table/values/Value.java b/table/src/main/java/tech/ydb/table/values/Value.java
index 91ed8df7..1c6a37e0 100644
--- a/table/src/main/java/tech/ydb/table/values/Value.java
+++ b/table/src/main/java/tech/ydb/table/values/Value.java
@@ -8,7 +8,7 @@
  * @author Sergey Polovko
  * @param <T> type of value
  */
-public interface Value<T extends Type> extends Serializable {
+public interface Value<T extends Type> extends Serializable, Comparable<Value<?>> {
 
     Value<?>[] EMPTY_ARRAY = {};
 
diff --git a/table/src/main/java/tech/ydb/table/values/VariantType.java b/table/src/main/java/tech/ydb/table/values/VariantType.java
index f885c077..4186266e 100644
--- a/table/src/main/java/tech/ydb/table/values/VariantType.java
+++ b/table/src/main/java/tech/ydb/table/values/VariantType.java
@@ -82,12 +82,12 @@ public int hashCode() {
     public String toString() {
         StringBuilder sb = new StringBuilder(128);
         sb.append("Variant<");
-        int count = getItemsCount();
+        int count = itemTypes.length;
         for (int i = 0; i < count; i++) {
-            sb.append(getItemType(i)).append(", ");
-        }
-        if (count != 0) {
-            sb.setLength(sb.length() - 1); // cut last comma
+            sb.append(itemTypes[i]);
+            if (i < count - 1) {
+                sb.append(", ");
+            }
         }
         sb.append('>');
         return sb.toString();
diff --git a/table/src/main/java/tech/ydb/table/values/VariantValue.java b/table/src/main/java/tech/ydb/table/values/VariantValue.java
index a8fe231b..7112aed4 100644
--- a/table/src/main/java/tech/ydb/table/values/VariantValue.java
+++ b/table/src/main/java/tech/ydb/table/values/VariantValue.java
@@ -70,4 +70,33 @@ public ValueProtos.Value toPb() {
         builder.setVariantIndex(typeIndex);
         return builder.build();
     }
+
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (!optional.isPresent()) {
+                throw new NullPointerException("Cannot compare value " + this + " with NULL");
+            }
+            return compareTo(optional.get());
+        }
+
+        if (!getType().equals(other.getType())) {
+            throw new IllegalArgumentException("Cannot compare value " + getType() + " with " + other.getType());
+        }
+
+        VariantValue variant = (VariantValue) other;
+
+        // Compare type indices first
+        int indexComparison = Integer.compare(typeIndex, variant.typeIndex);
+        if (indexComparison != 0) {
+            return indexComparison;
+        }
+
+        return item.compareTo(variant.item);
+    }
 }
diff --git a/table/src/main/java/tech/ydb/table/values/VoidType.java b/table/src/main/java/tech/ydb/table/values/VoidType.java
index 6c64334e..98c88c75 100644
--- a/table/src/main/java/tech/ydb/table/values/VoidType.java
+++ b/table/src/main/java/tech/ydb/table/values/VoidType.java
@@ -24,16 +24,6 @@ public Kind getKind() {
         return Kind.VOID;
     }
 
-    @Override
-    public boolean equals(Object o) {
-        return this == o;
-    }
-
-    @Override
-    public int hashCode() {
-        return 31 * Kind.VOID.hashCode();
-    }
-
     @Override
     public String toString() {
         return "Void";
diff --git a/table/src/main/java/tech/ydb/table/values/VoidValue.java b/table/src/main/java/tech/ydb/table/values/VoidValue.java
index ba96c879..496ab817 100644
--- a/table/src/main/java/tech/ydb/table/values/VoidValue.java
+++ b/table/src/main/java/tech/ydb/table/values/VoidValue.java
@@ -18,16 +18,6 @@ public static VoidValue of() {
         return INSTANCE;
     }
 
-    @Override
-    public boolean equals(Object o) {
-        return o == this;
-    }
-
-    @Override
-    public int hashCode() {
-        return 1987;
-    }
-
     @Override
     public String toString() {
         return "Void";
@@ -42,4 +32,26 @@ public VoidType getType() {
     public ValueProtos.Value toPb() {
         return ProtoValue.voidValue();
     }
+
+    @Override
+    public int compareTo(Value<?> other) {
+        if (other == null) {
+            throw new NullPointerException("Cannot compare with null value");
+        }
+
+        if (other instanceof OptionalValue) {
+            OptionalValue optional = (OptionalValue) other;
+            if (!optional.isPresent()) {
+                return 0;
+            }
+            return compareTo(optional.get());
+        }
+
+        if (other instanceof VoidValue || other instanceof NullValue) {
+            // All VoidValue and NullValue  are equal
+            return 0;
+        }
+
+        throw new IllegalArgumentException("Cannot compare value " + getType() + " with " + other.getType());
+    }
 }
diff --git a/table/src/main/java/tech/ydb/table/values/proto/ProtoValue.java b/table/src/main/java/tech/ydb/table/values/proto/ProtoValue.java
index 7eef37bb..14f6bcff 100644
--- a/table/src/main/java/tech/ydb/table/values/proto/ProtoValue.java
+++ b/table/src/main/java/tech/ydb/table/values/proto/ProtoValue.java
@@ -966,9 +966,11 @@ public String toString() {
         @Override
         public String getUuidString() {
             long hiBe = LittleEndian.bswap(high);
-            return
-                    digits(low, 8) + "-" + digits(low >>> 32, 4) + "-" + digits(low >>> 48, 4) + "-" +
-                            digits(hiBe >> 48, 4) + "-" + digits(hiBe, 12);
+            return digits(low, 8) + "-" +
+                    digits(low >>> 32, 4) + "-" +
+                    digits(low >>> 48, 4) + "-" +
+                    digits(hiBe >> 48, 4) + "-" +
+                    digits(hiBe, 12);
         }
 
         @Override
diff --git a/table/src/test/java/tech/ydb/table/integration/ValuesReadTest.java b/table/src/test/java/tech/ydb/table/integration/ValuesReadTest.java
index 00dba069..b8e31f2a 100644
--- a/table/src/test/java/tech/ydb/table/integration/ValuesReadTest.java
+++ b/table/src/test/java/tech/ydb/table/integration/ValuesReadTest.java
@@ -24,11 +24,14 @@
 import tech.ydb.table.transaction.TxControl;
 import tech.ydb.table.values.DecimalType;
 import tech.ydb.table.values.DecimalValue;
+import tech.ydb.table.values.ListValue;
 import tech.ydb.table.values.NullType;
 import tech.ydb.table.values.NullValue;
 import tech.ydb.table.values.PrimitiveType;
 import tech.ydb.table.values.PrimitiveValue;
+import tech.ydb.table.values.StructValue;
 import tech.ydb.table.values.Type;
+import tech.ydb.table.values.Value;
 import tech.ydb.test.junit4.GrpcTransportRule;
 
 /**
@@ -113,6 +116,89 @@ public void uuidReadTest() {
         Assert.assertEquals(0x6e73b41c4ede4d08L, v2.getUuidHigh());
     }
 
+    @Test
+    public void uuidSortTest() {
+        String[] sorted = new String[] {
+            "00000000-0000-0000-0000-000000000001",
+            "00000000-0000-0000-0000-000000000010",
+            "00000000-0000-0000-0000-000000000100",
+            "00000000-0000-0000-0000-000000001000",
+            "00000000-0000-0000-0000-000000010000",
+            "00000000-0000-0000-0000-000000100000",
+            "00000000-0000-0000-0000-000001000000",
+            "00000000-0000-0000-0000-000010000000",
+            "00000000-0000-0000-0000-000100000000",
+            "00000000-0000-0000-0000-001000000000",
+            "00000000-0000-0000-0000-010000000000",
+            "00000000-0000-0000-0000-100000000000",
+
+            "00000000-0000-0000-0001-000000000000",
+            "00000000-0000-0000-0010-000000000000",
+            "00000000-0000-0000-0100-000000000000",
+            "00000000-0000-0000-1000-000000000000",
+
+            "00000000-0000-0100-0000-000000000000",
+            "00000000-0000-1000-0000-000000000000",
+            "00000000-0000-0001-0000-000000000000",
+            "00000000-0000-0010-0000-000000000000",
+
+            "00000000-0100-0000-0000-000000000000",
+            "00000000-1000-0000-0000-000000000000",
+            "00000000-0001-0000-0000-000000000000",
+            "00000000-0010-0000-0000-000000000000",
+
+            "01000000-0000-0000-0000-000000000000",
+            "10000000-0000-0000-0000-000000000000",
+            "00010000-0000-0000-0000-000000000000",
+            "00100000-0000-0000-0000-000000000000",
+            "00000100-0000-0000-0000-000000000000",
+            "00001000-0000-0000-0000-000000000000",
+            "00000001-0000-0000-0000-000000000000",
+            "00000010-0000-0000-0000-000000000000",
+        };
+
+        StructValue[] sv = new StructValue[sorted.length];
+        for (int idx = 0; idx < sorted.length; idx++) {
+            sv[idx] = StructValue.of("uuid", PrimitiveValue.newUuid(sorted[idx]));
+        }
+        ListValue list = ListValue.of(sv);
+
+        DataQueryResult result = CTX.supplyResult(s -> s.executeDataQuery(""
+                + "DECLARE $input AS List<Struct<uuid: UUID>>;"
+                + "SELECT uuid FROM AS_TABLE($input) ORDER BY uuid ASC;"
+                + "SELECT uuid FROM AS_TABLE($input) ORDER BY uuid DESC;",
+                TxControl.snapshotRo(), Params.of("$input", list)
+        )).join().getValue();
+
+        Assert.assertEquals(2, result.getResultSetCount());
+        ResultSetReader rs1 = result.getResultSet(0);
+        ResultSetReader rs2 = result.getResultSet(1);
+
+        Value<?> p1 = null;
+        Value<?> p2 = null;
+        for (int idx = 0; idx < sorted.length; idx++) {
+            Assert.assertTrue(rs1.next());
+            Assert.assertTrue(rs2.next());
+
+            Assert.assertEquals(UUID.fromString(sorted[idx]), rs1.getColumn(0).getUuid());
+            Assert.assertEquals(UUID.fromString(sorted[sorted.length - 1 - idx]), rs2.getColumn(0).getUuid());
+
+            Value<?> v1 = rs1.getColumn(0).getValue();
+            Value<?> v2 = rs2.getColumn(0).getValue();
+
+            if (idx != 0) {
+                Assert.assertTrue("" + v1 + " > " + p1, v1.compareTo(p1) > 0);
+                Assert.assertTrue("" + v2 + " < " + p2, v2.compareTo(p2) < 0);
+            }
+
+            p1 = v1;
+            p2 = v2;
+        }
+
+        Assert.assertFalse(rs1.next());
+        Assert.assertFalse(rs2.next());
+    }
+
     private void assertTimestamp(ValueReader vr, boolean optional, Instant expected) {
         Assert.assertNotNull(vr);
         if (optional) {
diff --git a/table/src/test/java/tech/ydb/table/values/ValueComparableTest.java b/table/src/test/java/tech/ydb/table/values/ValueComparableTest.java
new file mode 100644
index 00000000..ad826c09
--- /dev/null
+++ b/table/src/test/java/tech/ydb/table/values/ValueComparableTest.java
@@ -0,0 +1,388 @@
+package tech.ydb.table.values;
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+
+/**
+ * Test for Comparable implementation of Value classes
+ */
+public class ValueComparableTest {
+
+    private <T> void assertNpe(String message, Comparable<T> one, T other) {
+        NullPointerException npe = Assert.assertThrows(NullPointerException.class, () -> one.compareTo(other));
+        Assert.assertEquals(message, npe.getMessage());
+    }
+
+    private <T> void assertIllegalArgument(String message, Comparable<T> one, T other) {
+        IllegalArgumentException ex = Assert.assertThrows(IllegalArgumentException.class, () -> one.compareTo(other));
+        Assert.assertEquals(message, ex.getMessage());
+    }
+
+    private <T> void assertLess(Comparable<T> one, T other) {
+        Assert.assertTrue("" + one + " < " + other + " FAILED", one.compareTo(other) < 0);
+    }
+
+    private <T> void assertGreater(Comparable<T> one, T other) {
+        Assert.assertTrue("" + one + " > " + other + " FAILED", one.compareTo(other) > 0);
+    }
+
+    private <T> void assertEquals(Comparable<T> one, T other) {
+        Assert.assertEquals("" + one + " = " + other + " FAILED", 0, one.compareTo(other));
+    }
+
+    @Test
+    public void testPrimitiveValueComparison() {
+        // Test numeric comparisons
+        PrimitiveValue int1 = PrimitiveValue.newInt32(1);
+        PrimitiveValue int2 = PrimitiveValue.newInt32(2);
+        PrimitiveValue int3 = PrimitiveValue.newInt32(1);
+
+        assertLess(int1, int2);
+        assertGreater(int2, int1);
+        assertEquals(int1, int3);
+
+        // Optional comparing
+        assertLess(int1.makeOptional(), int2);
+        assertGreater(int2, int1.makeOptional());
+        assertEquals(int1.makeOptional(), int3);
+        assertEquals(int1, int3.makeOptional());
+
+        // Invalid values
+        assertNpe("Cannot compare with null value", int1, null);
+
+        // Test string comparisons
+        PrimitiveValue text1 = PrimitiveValue.newText("abc");
+        PrimitiveValue text2 = PrimitiveValue.newText("def");
+        PrimitiveValue text3 = PrimitiveValue.newText("abc");
+
+        assertLess(text1, text2);
+        assertGreater(text2, text1);
+        assertEquals(text1, text3);
+
+        // Test boolean comparisons
+        PrimitiveValue bool1 = PrimitiveValue.newBool(false);
+        PrimitiveValue bool2 = PrimitiveValue.newBool(true);
+
+        assertLess(bool1, bool2);
+        assertGreater(bool2, bool1);
+
+        assertIllegalArgument("Cannot compare value Int32 with Text", int1, text1);
+        assertNpe("Cannot compare value 1 with NULL", int1, PrimitiveType.Int32.makeOptional().emptyValue());
+
+        // All types check
+        assertLess(PrimitiveValue.newInt8(Byte.MIN_VALUE), PrimitiveValue.newInt8(Byte.MAX_VALUE));
+        assertLess(PrimitiveValue.newInt16(Short.MIN_VALUE), PrimitiveValue.newInt16(Short.MAX_VALUE));
+        assertLess(PrimitiveValue.newInt32(Integer.MIN_VALUE), PrimitiveValue.newInt32(Integer.MAX_VALUE));
+        assertLess(PrimitiveValue.newInt64(Long.MIN_VALUE), PrimitiveValue.newInt64(Long.MAX_VALUE));
+
+        assertGreater(PrimitiveValue.newUint8(Byte.MIN_VALUE), PrimitiveValue.newUint8(Byte.MAX_VALUE));
+        assertGreater(PrimitiveValue.newUint16(Short.MIN_VALUE), PrimitiveValue.newUint16(Short.MAX_VALUE));
+        assertGreater(PrimitiveValue.newUint32(Integer.MIN_VALUE), PrimitiveValue.newUint32(Integer.MAX_VALUE));
+        assertGreater(PrimitiveValue.newUint64(Long.MIN_VALUE), PrimitiveValue.newUint64(Long.MAX_VALUE));
+
+        assertLess(PrimitiveValue.newFloat(1e-3f), PrimitiveValue.newFloat(1e-2f));
+        assertLess(PrimitiveValue.newDouble(1e-3d), PrimitiveValue.newDouble(1e-2d));
+
+        byte[] b1 = new byte[] { 0x01, 0x02 };
+        assertEquals(PrimitiveValue.newBytesOwn(b1), PrimitiveValue.newBytesOwn(b1));
+        assertEquals(PrimitiveValue.newBytes(b1), PrimitiveValue.newBytes(new byte[] { 0x01, 0x02 }));
+        assertLess(PrimitiveValue.newBytes(b1), PrimitiveValue.newBytes(new byte[] { 0x01, 0x02, 0x1 }));
+        assertLess(PrimitiveValue.newBytes(b1), PrimitiveValue.newBytes(new byte[] { 0x02 }));
+
+        assertLess(PrimitiveValue.newYson(b1), PrimitiveValue.newYson(new byte[] { 0x01, 0x03 }));
+
+        assertLess(PrimitiveValue.newText("abc"), PrimitiveValue.newText("abcd"));
+        assertLess(PrimitiveValue.newJson("['abc']"), PrimitiveValue.newJson("['abcd']"));
+        assertLess(PrimitiveValue.newJsonDocument("['abc']"), PrimitiveValue.newJsonDocument("['abcd']"));
+
+        for (String[] uuid : new String[][] {
+            // Sort by       3 2 1 0  5 4  7 6  8 9  A B C D E F
+            new String[] { "FFFFFFFE-FFFF-FFFF-FFFF-FFFFFFFFFFFF", "000000FF-0000-0000-0000-000000000000" },
+            new String[] { "FFFF3000-FFFF-FFFF-FFFF-FFFFFFFFFFFF", "00003100-0000-0000-0000-000000000000" },
+            new String[] { "FF390000-FFFF-FFFF-FFFF-FFFFFFFFFFFF", "00400000-0000-0000-0000-000000000000" },
+            new String[] { "00000000-FFFF-FFFF-FFFF-FFFFFFFFFFFF", "01000000-0000-0000-0000-000000000000" },
+
+            new String[] { "00000000-FFFE-FFFF-FFFF-FFFFFFFFFFFF", "00000000-00FF-0000-0000-000000000000" },
+            new String[] { "00000000-A000-FFFF-FFFF-FFFFFFFFFFFF", "00000000-A100-0000-0000-000000000000" },
+            new String[] { "00000000-0000-FF00-FFFF-FFFFFFFFFFFF", "00000000-0000-0001-0000-000000000000" },
+            new String[] { "00000000-0000-0200-FFFF-FFFFFFFFFFFF", "00000000-0000-2000-0000-000000000000" },
+
+            new String[] { "00000000-0000-0000-F0FF-FFFFFFFFFFFF", "00000000-0000-0000-F100-000000000000" },
+            new String[] { "00000000-0000-0000-00FE-FFFFFFFFFFFF", "00000000-0000-0000-00FF-000000000000" },
+
+            new String[] { "00000000-0000-0000-0000-FEFFFFFFFFFF", "00000000-0000-0000-0000-FF0000000000" },
+            new String[] { "00000000-0000-0000-0000-00ABFFFFFFFF", "00000000-0000-0000-0000-00AC00000000" },
+            new String[] { "00000000-0000-0000-0000-000050FFFFFF", "00000000-0000-0000-0000-000060000000" },
+            new String[] { "00000000-0000-0000-0000-00000045FFFF", "00000000-0000-0000-0000-000004600000" },
+            new String[] { "00000000-0000-0000-0000-0000000012FF", "00000000-0000-0000-0000-000000001300" },
+            new String[] { "00000000-0000-0000-0000-000000000000", "00000000-0000-0000-0000-000000000001" },
+        }) {
+            assertLess(PrimitiveValue.newUuid(uuid[0]), PrimitiveValue.newUuid(uuid[1]));
+        }
+
+        assertLess(PrimitiveValue.newDate(20000), PrimitiveValue.newDate(20001));
+        assertLess(PrimitiveValue.newDatetime(1728000000), PrimitiveValue.newDatetime(1728000001));
+        assertLess(
+                PrimitiveValue.newTimestamp(Instant.ofEpochSecond(1728000000, 123456000)),
+                PrimitiveValue.newTimestamp(Instant.ofEpochSecond(1728000000, 123457000))
+        );
+        assertLess(PrimitiveValue.newInterval(20000), PrimitiveValue.newInterval(20001));
+
+        assertLess(PrimitiveValue.newDate32(20000), PrimitiveValue.newDate32(20001));
+        assertLess(PrimitiveValue.newDatetime64(1728000000), PrimitiveValue.newDatetime64(1728000001));
+        assertLess(
+                PrimitiveValue.newTimestamp64(Instant.ofEpochSecond(1728000000, 123456000)),
+                PrimitiveValue.newTimestamp64(Instant.ofEpochSecond(1728000000, 123457000))
+        );
+        assertLess(PrimitiveValue.newInterval64(20000), PrimitiveValue.newInterval64(20001));
+    }
+
+    @Test
+    public void testListValueComparison() {
+        ListValue list1 = ListValue.of(PrimitiveValue.newInt32(1), PrimitiveValue.newInt32(2));
+        ListValue list2 = ListValue.of(
+                PrimitiveValue.newInt32(1).makeOptional(),
+                PrimitiveValue.newInt32(2).makeOptional()
+        );
+
+        assertEquals(list1, list2);
+        assertEquals(list1.makeOptional(), list2);
+        assertEquals(list1, list2.makeOptional());
+
+        ListValue list3 = ListValue.of(PrimitiveValue.newInt32(1), PrimitiveValue.newInt32(3));
+        ListValue list4 = ListValue.of(PrimitiveValue.newInt32(2), PrimitiveValue.newInt32(2));
+
+        assertLess(list1, list3);
+        assertLess(list2, list3);
+        assertGreater(list4, list3);
+        assertLess(list3, list4);
+
+        ListValue list5 = ListValue.of(PrimitiveValue.newText("A"), PrimitiveValue.newText("Z"));
+        ListValue list6 = ListValue.of(PrimitiveValue.newText("A"), PrimitiveValue.newText("Z"));
+        ListValue list7 = ListValue.of(PrimitiveValue.newText("A"));
+        ListValue list8 = ListValue.of(PrimitiveValue.newText("Z"));
+
+        assertEquals(list5, list6);
+        assertEquals(list6, list5);
+        assertLess(list7, list5); // shorter list comes first
+
+        // Test proper lexicographical ordering
+
+        // ('Z') should be "bigger" than ('A','Z') in lexicographical order
+        assertLess(list5, list8); // ('A','Z') < ('Z')
+        assertGreater(list8, list5); // ('Z') > ('A','Z')
+
+        assertNpe("Cannot compare with null value", list1, null);
+        assertIllegalArgument("Cannot compare value Int32 with Text", list1, list5);
+        assertIllegalArgument("Cannot compare value Int32 with Text", list2, list5);
+    }
+
+    @Test
+    public void testStructValueComparison() {
+        StructValue s1 = StructValue.of("a", PrimitiveValue.newInt32(1), "b", PrimitiveValue.newInt32(2));
+        StructValue s2 = StructValue.of("a", PrimitiveValue.newInt32(1), "b", PrimitiveValue.newInt32(2));
+        StructValue s3 = StructValue.of("a", PrimitiveValue.newInt32(2), "b", PrimitiveValue.newInt32(1));
+        StructValue s4 = StructValue.of("a", PrimitiveValue.newInt32(1), "b", PrimitiveValue.newText("a"));
+        StructValue s5 = StructValue.of("a", PrimitiveValue.newInt32(1));
+
+        assertEquals(s1, s2);
+        assertEquals(s1, s2.makeOptional());
+        assertEquals(s1.makeOptional(), s2);
+
+        assertLess(s1, s3);
+        assertGreater(s3, s1);
+
+        assertNpe("Cannot compare with null value", s1, null);
+        assertIllegalArgument("Cannot compare value Struct<'a': Int32, 'b': Int32> with Struct<'a': Int32, 'b': Text>",
+                s1, s4);
+        assertIllegalArgument("Cannot compare value Struct<'a': Int32, 'b': Int32> with Struct<'a': Int32>", s1, s5);
+    }
+
+    @Test
+    public void testDictValueComparison() {
+        DictValue d1 = DictValue.of(PrimitiveValue.newText("a"), PrimitiveValue.newInt32(1));
+        DictValue d2 = DictValue.of(PrimitiveValue.newText("a"), PrimitiveValue.newInt32(2));
+        DictValue d3 = DictValue.of(PrimitiveValue.newText("a"), PrimitiveValue.newInt32(1));
+
+        assertLess(d1, d2);
+        assertGreater(d2, d1);
+
+        assertEquals(d1, d3);
+        assertEquals(d1, d3.makeOptional());
+        assertEquals(d1.makeOptional(), d3);
+
+        DictValue d4 = DictValue.of(PrimitiveValue.newText("a"), PrimitiveValue.newText("abc"));
+        assertIllegalArgument("Cannot compare value Dict<Text, Int32> with Dict<Text, Text>", d1, d4);
+        assertIllegalArgument("Cannot compare value Dict<Text, Text> with Dict<Text, Int32>", d4, d1);
+
+        DictValue d5 = DictValue.of(PrimitiveValue.newText("b"), PrimitiveValue.newInt32(1));
+
+        // {"a": 1} should be "bigger" than {"b": 1 } in lexicographical order
+        assertGreater(d1, d5);
+        assertLess(d5, d1);
+
+        Map<Value<?>, Value<?>> map6 = new HashMap<>();
+        map6.put(PrimitiveValue.newText("a"), PrimitiveValue.newInt32(1));
+        map6.put(PrimitiveValue.newText("b"), PrimitiveValue.newInt32(2));
+        DictValue d6 = DictType.of(PrimitiveType.Text, PrimitiveType.Int32).newValueOwn(map6);
+
+        assertLess(d1, d6); // {"a": 1} < {"a": 1, "b": 2} (prefix case)
+        assertGreater(d6, d1); // {"a": 1, "b": 2} > {"a": 1}
+    }
+
+    @Test
+    public void testOptionalValueComparison() {
+        OptionalValue opt1 = OptionalValue.of(PrimitiveValue.newInt32(1));
+        OptionalValue opt2 = OptionalValue.of(PrimitiveValue.newInt32(2));
+        OptionalValue opt3 = OptionalValue.of(PrimitiveValue.newInt32(1));
+        OptionalValue opt4 = PrimitiveType.Int32.makeOptional().emptyValue();
+        OptionalValue opt5 = OptionalValue.of(PrimitiveValue.newText("abc"));
+        OptionalValue opt6 = PrimitiveType.Text.makeOptional().emptyValue();
+
+        assertLess(opt1, opt2);
+        assertGreater(opt2, opt1);
+        assertEquals(opt1, opt3);
+
+        assertNpe("Cannot compare NULL with value 1", opt4, opt1);
+        assertNpe("Cannot compare value 1 with NULL", opt1, opt4);
+
+        assertIllegalArgument("Cannot compare value Int32 with Text", opt1, opt5);
+        assertIllegalArgument("Cannot compare value Text with Int32", opt5, opt1);
+
+        assertEquals(opt4, opt6);
+        assertEquals(opt6, opt4);
+
+        assertNpe("Cannot compare with null value", opt1, null);
+    }
+
+    @Test
+    public void testTupleValueComparison() {
+        TupleValue t1 = TupleValue.of(PrimitiveValue.newInt32(1), PrimitiveValue.newInt32(2));
+        TupleValue t2 = TupleValue.of(PrimitiveValue.newInt32(1), PrimitiveValue.newInt32(3));
+        TupleValue t3 = TupleValue.of(PrimitiveValue.newInt32(1), PrimitiveValue.newInt32(2));
+        TupleValue t4 = TupleValue.of(PrimitiveValue.newInt32(1));
+        TupleValue t5 = TupleValue.of(PrimitiveValue.newInt32(1), PrimitiveValue.newUint32(2));
+
+        assertLess(t1, t2);
+        assertGreater(t2, t1);
+        assertEquals(t1, t3);
+        assertEquals(t1, t3.makeOptional());
+        assertEquals(t1.makeOptional(), t3);
+
+        assertNpe("Cannot compare with null value", t1, null);
+        assertIllegalArgument("Cannot compare value Tuple<Int32> with Tuple<Int32, Int32>", t4, t1);
+        assertIllegalArgument("Cannot compare value Tuple<Int32, Uint32> with Tuple<Int32, Int32>", t5, t1);
+    }
+
+    @Test
+    public void testVariantValueComparison() {
+        VariantType t1 = VariantType.ofOwn(PrimitiveType.Int32, PrimitiveType.Text);
+        VariantType t2 = VariantType.ofOwn(PrimitiveType.Int32, PrimitiveType.Text, PrimitiveType.Int32);
+
+        VariantValue v1 = new VariantValue(t1, PrimitiveValue.newInt32(1), 0);
+        VariantValue v2 = new VariantValue(t1, PrimitiveValue.newText("abc"), 1);
+        VariantValue v3 = new VariantValue(t1, PrimitiveValue.newInt32(1), 0);
+        VariantValue v4 = new VariantValue(t1, PrimitiveValue.newText("aBc"), 1);
+        VariantValue v5 = new VariantValue(t2, PrimitiveValue.newInt32(1), 0);
+
+        assertLess(v1, v2); // type index 0 < 1
+        assertGreater(v2, v1);
+        assertEquals(v1, v3);
+        assertEquals(v3, v1);
+        assertGreater(v2, v4.makeOptional());
+        assertLess(v4.makeOptional(), v2);
+
+        assertNpe("Cannot compare with null value", v1, null);
+        assertNpe("Cannot compare value Variant[0; 1] with NULL", v1, t1.makeOptional().emptyValue());
+        assertNpe("Cannot compare NULL with value Variant[0; 1]", t1.makeOptional().emptyValue(), v1);
+
+        assertIllegalArgument("Cannot compare value Variant<Int32, Text> with Variant<Int32, Text, Int32>", v1, v5);
+        assertIllegalArgument("Cannot compare value Variant<Int32, Text, Int32> with Variant<Int32, Text>", v5, v1);
+    }
+
+    @Test
+    public void testVoidValueComparison() {
+        VoidValue void1 = VoidValue.of();
+        VoidValue void2 = VoidValue.of();
+
+        assertEquals(void1, void2);
+        assertEquals(void1, void2.makeOptional());
+        assertEquals(void1.makeOptional(), void2);
+
+        assertEquals(void1, NullValue.of());
+        assertEquals(void1, NullValue.of().makeOptional());
+        assertEquals(void1, PrimitiveType.Int32.makeOptional().emptyValue());
+
+        assertNpe("Cannot compare with null value", void1, null);
+        assertIllegalArgument("Cannot compare value Void with Int32", void1, PrimitiveValue.newInt32(1));
+    }
+
+    @Test
+    public void testNullValueComparison() {
+        NullValue null1 = NullValue.of();
+        NullValue null2 = NullValue.of();
+
+        assertEquals(null1, null2);
+        assertEquals(null1, null2.makeOptional());
+        assertEquals(null1.makeOptional(), null2);
+
+        assertEquals(null1, VoidValue.of());
+        assertEquals(null1, VoidValue.of().makeOptional());
+        assertEquals(null1, PrimitiveType.Int32.makeOptional().emptyValue());
+
+        assertNpe("Cannot compare with null value", null1, null);
+        assertIllegalArgument("Cannot compare value Null with Int32", null1, PrimitiveValue.newInt32(1));
+    }
+
+    @Test
+    public void testDecimalValueComparison() {
+        DecimalType t1 = DecimalType.of(33, 2);
+        DecimalType t2 = DecimalType.of(30, 9);
+        DecimalType t3 = DecimalType.of(11, 2);
+
+        assertEquals(t1.newValue("1"), t2.newValue("1"));
+        assertEquals(t1.newValue("1.23").makeOptional(), t2.newValue("1.23"));
+        assertEquals(t1.newValue("-1.23"), t2.newValue("-1.23").makeOptional());
+
+        // the same scale
+        assertEquals(t3.newValue("999999999.99"), t1.newValue("999999999.99"));
+        assertEquals(t3.newValue("-999999999.99"), t1.newValue("-999999999.99"));
+        assertLess(t3.newValue("999999999.98"), t1.newValue("999999999.99"));
+        assertLess(t3.newValue("899999999.99"), t1.newValue("999999999.99"));
+        assertGreater(t3.newValue("-999999999.98"), t1.newValue("-999999999.99"));
+        assertGreater(t3.newValue("-899999999.99"), t1.newValue("-999999999.99"));
+
+        // the differnt scales
+        assertEquals(t3.newValue("999999999.99"), t2.newValue("999999999.99"));
+        assertEquals(t3.newValue("-999999999.99"), t2.newValue("-999999999.99"));
+        assertLess(t3.newValue("999999999.99"), t2.newValue("1000000000"));
+        assertGreater(t3.newValue("-999999999.99"), t2.newValue("-1000000000"));
+        assertLess(t1.newValue("1.23"), t2.newValue("1.234"));
+        assertGreater(t1.newValue("-1.23"), t2.newValue("-1.234"));
+
+        // special values
+        assertEquals(t1.getInf(), t2.getInf());
+        assertEquals(t1.getNegInf(), t2.getNegInf());
+        assertEquals(t1.getNaN(), t2.getNaN());
+
+        // type bounds
+        assertLess(t1.getNegInf(), t2.newValue("-999999999999999999999.999999999"));
+        assertEquals(t1.getNegInf(), t2.newValue("-1000000000000000000000"));
+
+        assertGreater(t1.getInf(), t2.newValue("999999999999999999999.999999999"));
+        assertEquals(t1.getInf(), t2.newValue("1000000000000000000000"));
+
+        assertGreater(t1.getNaN(), t2.newValue("999999999999999999999.999999999"));
+        assertGreater(t1.getNaN(), t2.newValue("9000000000000000000000"));
+
+        // errors
+        assertNpe("Cannot compare with null value", t1.newValue("1"), null);
+        assertNpe("Cannot compare value 1.00 with NULL", t1.newValue("1"), t1.makeOptional().emptyValue());
+        assertIllegalArgument("Cannot compare DecimalValue with Int32", t1.newValue("1"), PrimitiveValue.newInt32(1));
+    }
+}
\ No newline at end of file
diff --git a/topic/src/main/java/tech/ydb/topic/write/WriteAck.java b/topic/src/main/java/tech/ydb/topic/write/WriteAck.java
index b7b082f1..7d8b2107 100644
--- a/topic/src/main/java/tech/ydb/topic/write/WriteAck.java
+++ b/topic/src/main/java/tech/ydb/topic/write/WriteAck.java
@@ -1,5 +1,7 @@
 package tech.ydb.topic.write;
 
+import java.time.Duration;
+
 /**
  * @author Nikolay Perfilov
  */
@@ -7,11 +9,13 @@ public class WriteAck {
     private final long seqNo;
     private final State state;
     private final Details details;
+    private final Statistics statistics;
 
-    public WriteAck(long seqNo, State state, Details details) {
+    public WriteAck(long seqNo, State state, Details details, Statistics statistics) {
         this.seqNo = seqNo;
         this.state = state;
         this.details = details;
+        this.statistics = statistics;
     }
 
     public enum State {
@@ -20,18 +24,6 @@ public enum State {
         WRITTEN_IN_TX
     }
 
-    public static class Details {
-        private final long offset;
-
-        public Details(long offset) {
-            this.offset = offset;
-        }
-
-        public long getOffset() {
-            return offset;
-        }
-    }
-
     public long getSeqNo() {
         return seqNo;
     }
@@ -47,4 +39,95 @@ public State getState() {
     public Details getDetails() {
         return details;
     }
+
+    /**
+     * Returns write statistics associated with this write confirmation.
+     * Note: The statistics may cover multiple messages confirmed together by the server.
+     * Although this WriteAck corresponds to a single written message, the server may confirm several messages in a single response.
+     * Therefore, the returned statistics may represent the combined data for all messages included in the same write confirmation from the server.
+     * @return {@link Statistics} with timings if statistics are available or null otherwise
+     */
+    public Statistics getStatistics() {
+        return statistics;
+    }
+
+    public static class Details {
+        private final long offset;
+
+        public Details(long offset) {
+            this.offset = offset;
+        }
+
+        public long getOffset() {
+            return offset;
+        }
+    }
+
+    /**
+     * Messages batch statistics.
+     * All messages within the batch are persisted together so write
+     * statistics is for the whole messages batch.
+     */
+    public static class Statistics {
+        private final Duration persistingTime;
+        private final Duration partitionQuotaWaitTime;
+        private final Duration topicQuotaWaitTime;
+        private final Duration maxQueueWaitTime;
+        private final Duration minQueueWaitTime;
+
+        /**
+         * Create the messages batch statistics object, for a single messages batch.
+         *
+         * @param persistingTime
+         * @param partitionQuotaWaitTime
+         * @param topicQuotaWaitTime
+         * @param maxQueueWaitTime
+         * @param minQueueWaitTime
+         */
+        public Statistics(Duration persistingTime,
+                Duration partitionQuotaWaitTime, Duration topicQuotaWaitTime,
+                Duration maxQueueWaitTime, Duration minQueueWaitTime) {
+            this.persistingTime = persistingTime;
+            this.partitionQuotaWaitTime = partitionQuotaWaitTime;
+            this.topicQuotaWaitTime = topicQuotaWaitTime;
+            this.maxQueueWaitTime = maxQueueWaitTime;
+            this.minQueueWaitTime = minQueueWaitTime;
+        }
+
+        /**
+         * @return Time spent in persisting of data.
+         */
+        public Duration getPersistingTime() {
+            return persistingTime;
+        }
+
+        /**
+         * @return Time spent awaiting for partition write quota.
+         */
+        public Duration getPartitionQuotaWaitTime() {
+            return partitionQuotaWaitTime;
+        }
+
+        /**
+         * @return Time spent awaiting for topic write quota.
+         */
+        public Duration getTopicQuotaWaitTime() {
+            return topicQuotaWaitTime;
+        }
+
+        /**
+         * @return Time spent in queue before persisting, maximal of all messages in response.
+         */
+        public Duration getMaxQueueWaitTime() {
+            return maxQueueWaitTime;
+        }
+
+        /**
+         * @return Time spent in queue before persisting, minimal of all messages in response.
+         */
+        public Duration getMinQueueWaitTime() {
+            return minQueueWaitTime;
+        }
+    }
+
 }
diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java
index 58733d67..7617fe9f 100644
--- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java
+++ b/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java
@@ -20,6 +20,7 @@
 import tech.ydb.core.Issue;
 import tech.ydb.core.Status;
 import tech.ydb.core.StatusCode;
+import tech.ydb.core.utils.ProtobufUtils;
 import tech.ydb.proto.StatusCodesProtos;
 import tech.ydb.proto.topic.YdbTopic;
 import tech.ydb.topic.TopicRpc;
@@ -421,6 +422,17 @@ private void onInitResponse(YdbTopic.StreamWriteMessage.InitResponse response) {
         private void onWriteResponse(YdbTopic.StreamWriteMessage.WriteResponse response) {
             List<YdbTopic.StreamWriteMessage.WriteResponse.WriteAck> acks = response.getAcksList();
             logger.debug("[{}] Received WriteResponse with {} WriteAcks", streamId, acks.size());
+            WriteAck.Statistics statistics = null;
+            if (response.getWriteStatistics() != null) {
+                YdbTopic.StreamWriteMessage.WriteResponse.WriteStatistics src = response.getWriteStatistics();
+                statistics = new WriteAck.Statistics(
+                    ProtobufUtils.protoToDuration(src.getPersistingTime()),
+                    ProtobufUtils.protoToDuration(src.getPartitionQuotaWaitTime()),
+                    ProtobufUtils.protoToDuration(src.getTopicQuotaWaitTime()),
+                    ProtobufUtils.protoToDuration(src.getMaxQueueWaitTime()),
+                    ProtobufUtils.protoToDuration(src.getMinQueueWaitTime())
+                );
+            }
             int inFlightFreed = 0;
             long bytesFreed = 0;
             for (YdbTopic.StreamWriteMessage.WriteResponse.WriteAck ack : acks) {
@@ -433,7 +445,7 @@ private void onWriteResponse(YdbTopic.StreamWriteMessage.WriteResponse response)
                         inFlightFreed++;
                         bytesFreed += sentMessage.getSize();
                         sentMessages.remove();
-                        processWriteAck(sentMessage, ack);
+                        processWriteAck(sentMessage, statistics, ack);
                         break;
                     }
                     if (sentMessage.getSeqNo() < ack.getSeqNo()) {
@@ -474,7 +486,7 @@ private void processMessage(YdbTopic.StreamWriteMessage.FromServer message) {
             }
         }
 
-        private void processWriteAck(EnqueuedMessage message,
+        private void processWriteAck(EnqueuedMessage message, WriteAck.Statistics statistics,
                                      YdbTopic.StreamWriteMessage.WriteResponse.WriteAck ack) {
             logger.debug("[{}] Received WriteAck with seqNo {} and status {}", streamId, ack.getSeqNo(),
                     ack.getMessageWriteStatusCase());
@@ -482,12 +494,12 @@ private void processWriteAck(EnqueuedMessage message,
             switch (ack.getMessageWriteStatusCase()) {
                 case WRITTEN:
                     WriteAck.Details details = new WriteAck.Details(ack.getWritten().getOffset());
-                    resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.WRITTEN, details);
+                    resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.WRITTEN, details, statistics);
                     break;
                 case SKIPPED:
                     switch (ack.getSkipped().getReason()) {
                         case REASON_ALREADY_WRITTEN:
-                            resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.ALREADY_WRITTEN, null);
+                            resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.ALREADY_WRITTEN, null, statistics);
                             break;
                         case REASON_UNSPECIFIED:
                         default:
@@ -497,7 +509,7 @@ private void processWriteAck(EnqueuedMessage message,
                     }
                     break;
                 case WRITTEN_IN_TX:
-                    resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.WRITTEN_IN_TX, null);
+                    resultAck = new WriteAck(ack.getSeqNo(), WriteAck.State.WRITTEN_IN_TX, null, statistics);
                     break;
                 default:
                     message.getFuture().completeExceptionally(
@@ -519,5 +531,6 @@ private void closeDueToError(Status status, Throwable th) {
         protected void onStop() {
             logger.debug("[{}] Session {} onStop called", streamId, sessionId);
         }
+
     }
 }