diff --git a/CHANGELOG.md b/CHANGELOG.md
index dfc9179da3..cf37896a28 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Add experimental composite samplers
+  ([#4714](https://github.com/open-telemetry/opentelemetry-python/pull/4714))
+
 ## Version 1.36.0/0.57b0 (2025-07-29)
 
 - Add missing Prometheus exporter documentation
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/__init__.py
new file mode 100644
index 0000000000..1a8c372276
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/__init__.py
@@ -0,0 +1,31 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+__all__ = [
+    "ComposableSampler",
+    "SamplingIntent",
+    "composable_always_off",
+    "composable_always_on",
+    "composable_parent_threshold",
+    "composable_traceid_ratio_based",
+    "composite_sampler",
+]
+
+
+from ._always_off import composable_always_off
+from ._always_on import composable_always_on
+from ._composable import ComposableSampler, SamplingIntent
+from ._parent_threshold import composable_parent_threshold
+from ._sampler import composite_sampler
+from ._traceid_ratio import composable_traceid_ratio_based
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_always_off.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_always_off.py
new file mode 100644
index 0000000000..eaafe16416
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_always_off.py
@@ -0,0 +1,55 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from __future__ import annotations
+
+from typing import Sequence
+
+from opentelemetry.context import Context
+from opentelemetry.trace import Link, SpanKind, TraceState
+from opentelemetry.util.types import Attributes
+
+from ._composable import ComposableSampler, SamplingIntent
+from ._util import INVALID_THRESHOLD
+
+_intent = SamplingIntent(threshold=INVALID_THRESHOLD, threshold_reliable=False)
+
+
+class _ComposableAlwaysOffSampler(ComposableSampler):
+    def sampling_intent(
+        self,
+        parent_ctx: Context | None,
+        name: str,
+        span_kind: SpanKind | None,
+        attributes: Attributes,
+        links: Sequence[Link] | None,
+        trace_state: TraceState | None = None,
+    ) -> SamplingIntent:
+        return _intent
+
+    def get_description(self) -> str:
+        return "ComposableAlwaysOff"
+
+
+_always_off = _ComposableAlwaysOffSampler()
+
+
+def composable_always_off() -> ComposableSampler:
+    """Returns a composable sampler that does not sample any span.
+
+    - Always returns a SamplingIntent with no threshold, indicating all spans should be dropped
+    - Sets threshold_reliable to false
+    - Does not add any attributes
+    """
+    return _always_off
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_always_on.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_always_on.py
new file mode 100644
index 0000000000..88ac61c5d3
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_always_on.py
@@ -0,0 +1,55 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from __future__ import annotations
+
+from typing import Sequence
+
+from opentelemetry.context import Context
+from opentelemetry.trace import Link, SpanKind, TraceState
+from opentelemetry.util.types import Attributes
+
+from ._composable import ComposableSampler, SamplingIntent
+from ._util import MIN_THRESHOLD
+
+_intent = SamplingIntent(threshold=MIN_THRESHOLD)
+
+
+class _ComposableAlwaysOnSampler(ComposableSampler):
+    def sampling_intent(
+        self,
+        parent_ctx: Context | None,
+        name: str,
+        span_kind: SpanKind | None,
+        attributes: Attributes,
+        links: Sequence[Link] | None,
+        trace_state: TraceState | None = None,
+    ) -> SamplingIntent:
+        return _intent
+
+    def get_description(self) -> str:
+        return "ComposableAlwaysOn"
+
+
+_always_on = _ComposableAlwaysOnSampler()
+
+
+def composable_always_on() -> ComposableSampler:
+    """Returns a composable sampler that samples all spans.
+
+    - Always returns a SamplingIntent with threshold set to sample all spans (threshold = 0)
+    - Sets threshold_reliable to true
+    - Does not add any attributes
+    """
+    return _always_on
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_composable.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_composable.py
new file mode 100644
index 0000000000..80c0462f9b
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_composable.py
@@ -0,0 +1,59 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Callable, Protocol, Sequence
+
+from opentelemetry.context import Context
+from opentelemetry.trace import Link, SpanKind, TraceState
+from opentelemetry.util.types import Attributes
+
+
+@dataclass(frozen=True)
+class SamplingIntent:
+    """Information to make a consistent sampling decision."""
+
+    threshold: int
+    """The sampling threshold value. A lower threshold increases the likelihood of sampling."""
+
+    threshold_reliable: bool = field(default=True)
+    """Indicates whether the threshold is reliable for Span-to-Metrics estimation."""
+
+    attributes: Attributes = field(default=None)
+    """Any attributes to be added to a sampled span."""
+
+    update_trace_state: Callable[[TraceState], TraceState] = field(
+        default=lambda ts: ts
+    )
+    """Any updates to be made to trace state."""
+
+
+class ComposableSampler(Protocol):
+    """A sampler that can be composed to make a final sampling decision."""
+
+    def sampling_intent(
+        self,
+        parent_ctx: Context | None,
+        name: str,
+        span_kind: SpanKind | None,
+        attributes: Attributes,
+        links: Sequence[Link] | None,
+        trace_state: TraceState | None,
+    ) -> SamplingIntent:
+        """Returns information to make a sampling decision."""
+
+    def get_description(self) -> str:
+        """Returns a description of the sampler."""
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_parent_threshold.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_parent_threshold.py
new file mode 100644
index 0000000000..83b7b7d300
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_parent_threshold.py
@@ -0,0 +1,89 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from __future__ import annotations
+
+from typing import Sequence
+
+from opentelemetry.context import Context
+from opentelemetry.trace import Link, SpanKind, TraceState, get_current_span
+from opentelemetry.util.types import Attributes
+
+from ._composable import ComposableSampler, SamplingIntent
+from ._trace_state import OtelTraceState
+from ._util import (
+    INVALID_THRESHOLD,
+    MIN_THRESHOLD,
+    is_valid_threshold,
+)
+
+
+class _ComposableParentThreshold(ComposableSampler):
+    def __init__(self, root_sampler: ComposableSampler):
+        self._root_sampler = root_sampler
+        self._description = f"ComposableParentThreshold{{root={root_sampler.get_description()}}}"
+
+    def sampling_intent(
+        self,
+        parent_ctx: Context | None,
+        name: str,
+        span_kind: SpanKind | None,
+        attributes: Attributes,
+        links: Sequence[Link] | None,
+        trace_state: TraceState | None = None,
+    ) -> SamplingIntent:
+        parent_span = get_current_span(parent_ctx)
+        parent_span_ctx = parent_span.get_span_context()
+        is_root = not parent_span_ctx.is_valid
+        if is_root:
+            return self._root_sampler.sampling_intent(
+                parent_ctx, name, span_kind, attributes, links, trace_state
+            )
+
+        ot_trace_state = OtelTraceState.parse(trace_state)
+
+        if is_valid_threshold(ot_trace_state.threshold):
+            return SamplingIntent(
+                threshold=ot_trace_state.threshold,
+                threshold_reliable=True,
+            )
+
+        threshold = (
+            MIN_THRESHOLD
+            if parent_span_ctx.trace_flags.sampled
+            else INVALID_THRESHOLD
+        )
+        return SamplingIntent(threshold=threshold, threshold_reliable=False)
+
+    def get_description(self) -> str:
+        return self._description
+
+
+def composable_parent_threshold(
+    root_sampler: ComposableSampler,
+) -> ComposableSampler:
+    """Returns a consistent sampler that respects the sampling decision of
+    the parent span or falls-back to the given sampler if it is a root span.
+
+    - For spans without a parent context, delegate to the root sampler
+    - For spans with a parent context, returns a SamplingIntent that propagates the parent's sampling decision
+    - Returns the parent's threshold if available; otherwise, if the parent's sampled flag is set,
+    returns threshold=0; otherwise, if the parent's sampled flag is not set, no threshold is returned.
+    - Sets threshold_reliable to match the parent’s reliability, which is true if the parent had a threshold.
+    - Does not add any attributes
+
+    Args:
+        root_sampler: The root sampler to use for spans without a parent context.
+    """
+    return _ComposableParentThreshold(root_sampler)
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_sampler.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_sampler.py
new file mode 100644
index 0000000000..989cc36019
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_sampler.py
@@ -0,0 +1,101 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from __future__ import annotations
+
+from typing import Sequence
+
+from opentelemetry.context import Context
+from opentelemetry.sdk.trace.sampling import Decision, Sampler, SamplingResult
+from opentelemetry.trace import Link, SpanKind, TraceState
+from opentelemetry.util.types import Attributes
+
+from ._composable import ComposableSampler, SamplingIntent
+from ._trace_state import OTEL_TRACE_STATE_KEY, OtelTraceState
+from ._util import INVALID_THRESHOLD, is_valid_random_value, is_valid_threshold
+
+
+class _CompositeSampler(Sampler):
+    def __init__(self, delegate: ComposableSampler):
+        self._delegate = delegate
+
+    def should_sample(
+        self,
+        parent_context: Context | None,
+        trace_id: int,
+        name: str,
+        kind: SpanKind | None = None,
+        attributes: Attributes | None = None,
+        links: Sequence[Link] | None = None,
+        trace_state: TraceState | None = None,
+    ) -> SamplingResult:
+        ot_trace_state = OtelTraceState.parse(trace_state)
+
+        intent = self._delegate.sampling_intent(
+            parent_context, name, kind, attributes, links, trace_state
+        )
+        threshold = intent.threshold
+
+        if is_valid_threshold(threshold):
+            adjusted_count_correct = intent.threshold_reliable
+            if is_valid_random_value(ot_trace_state.random_value):
+                randomness = ot_trace_state.random_value
+            else:
+                # Use last 56 bits of trace_id as randomness
+                randomness = trace_id & 0x00FFFFFFFFFFFFFF
+            sampled = threshold <= randomness
+        else:
+            sampled = False
+            adjusted_count_correct = False
+
+        decision = Decision.RECORD_AND_SAMPLE if sampled else Decision.DROP
+        if sampled and adjusted_count_correct:
+            ot_trace_state.threshold = threshold
+        else:
+            ot_trace_state.threshold = INVALID_THRESHOLD
+
+        return SamplingResult(
+            decision,
+            intent.attributes,
+            _update_trace_state(trace_state, ot_trace_state, intent),
+        )
+
+    def get_description(self) -> str:
+        return self._delegate.get_description()
+
+
+def _update_trace_state(
+    trace_state: TraceState | None,
+    ot_trace_state: OtelTraceState,
+    intent: SamplingIntent,
+) -> TraceState | None:
+    otts = ot_trace_state.serialize()
+    if not trace_state:
+        if otts:
+            return TraceState(((OTEL_TRACE_STATE_KEY, otts),))
+        return None
+    new_trace_state = intent.update_trace_state(trace_state)
+    if otts:
+        return new_trace_state.update(OTEL_TRACE_STATE_KEY, otts)
+    return new_trace_state
+
+
+def composite_sampler(delegate: ComposableSampler) -> Sampler:
+    """A sampler that uses a a composable sampler to make its decision while
+    handling tracestate.
+
+    Args:
+        delegate: The composable sampler to use for making sampling decisions.
+    """
+    return _CompositeSampler(delegate)
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_trace_state.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_trace_state.py
new file mode 100644
index 0000000000..79d0cee413
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_trace_state.py
@@ -0,0 +1,141 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Sequence
+
+from opentelemetry.trace import TraceState
+
+from ._util import (
+    INVALID_RANDOM_VALUE,
+    INVALID_THRESHOLD,
+    MAX_THRESHOLD,
+    is_valid_random_value,
+    is_valid_threshold,
+)
+
+OTEL_TRACE_STATE_KEY = "ot"
+
+_TRACE_STATE_SIZE_LIMIT = 256
+_MAX_VALUE_LENGTH = 14  # 56 bits, 4 bits per hex digit
+
+
+@dataclass
+class OtelTraceState:
+    """Marshals OpenTelemetry tracestate for sampling parameters.
+
+    https://opentelemetry.io/docs/specs/otel/trace/tracestate-probability-sampling/
+    """
+
+    random_value: int
+    threshold: int
+    rest: Sequence[str]
+
+    @staticmethod
+    def invalid() -> "OtelTraceState":
+        return OtelTraceState(INVALID_RANDOM_VALUE, INVALID_THRESHOLD, ())
+
+    @staticmethod
+    def parse(trace_state: TraceState | None) -> "OtelTraceState":
+        if not trace_state:
+            return OtelTraceState.invalid()
+
+        ot = trace_state.get(OTEL_TRACE_STATE_KEY, "")
+
+        if not ot or len(ot) > _TRACE_STATE_SIZE_LIMIT:
+            return OtelTraceState.invalid()
+
+        threshold = INVALID_THRESHOLD
+        random_value = INVALID_RANDOM_VALUE
+
+        members = ot.split(";")
+        rest: list[str] | None = None
+        for member in members:
+            if member.startswith("th:"):
+                threshold = _parse_th(member[len("th:") :], INVALID_THRESHOLD)
+                continue
+            if member.startswith("rv:"):
+                random_value = _parse_rv(
+                    member[len("rv:") :], INVALID_RANDOM_VALUE
+                )
+                continue
+            if rest is None:
+                rest = [member]
+            else:
+                rest.append(member)
+
+        return OtelTraceState(
+            random_value=random_value, threshold=threshold, rest=rest or ()
+        )
+
+    def serialize(self) -> str:
+        if (
+            not is_valid_threshold(self.threshold)
+            and not is_valid_random_value(self.random_value)
+            and not self.rest
+        ):
+            return ""
+
+        parts: list[str] = []
+        if (
+            is_valid_threshold(self.threshold)
+            and self.threshold != MAX_THRESHOLD
+        ):
+            parts.append(f"th:{serialize_th(self.threshold)}")
+        if is_valid_random_value(self.random_value):
+            parts.append(f"rv:{_serialize_rv(self.random_value)}")
+        if self.rest:
+            parts.extend(self.rest)
+        res = ";".join(parts)
+        while len(res) > _TRACE_STATE_SIZE_LIMIT:
+            delim_idx = res.rfind(";")
+            if delim_idx == -1:
+                break
+            res = res[:delim_idx]
+        return res
+
+
+def _parse_th(value: str, default: int) -> int:
+    if not value or len(value) > _MAX_VALUE_LENGTH:
+        return default
+
+    try:
+        parsed = int(value, 16)
+    except ValueError:
+        return default
+
+    trailing_zeros = _MAX_VALUE_LENGTH - len(value)
+    return parsed << (trailing_zeros * 4)
+
+
+def _parse_rv(value: str, default: int) -> int:
+    if not value or len(value) != _MAX_VALUE_LENGTH:
+        return default
+
+    try:
+        return int(value, 16)
+    except ValueError:
+        return default
+
+
+def serialize_th(threshold: int) -> str:
+    if not threshold:
+        return "0"
+    return f"{threshold:014x}".rstrip("0")
+
+
+def _serialize_rv(random_value: int) -> str:
+    return f"{random_value:014x}"
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_traceid_ratio.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_traceid_ratio.py
new file mode 100644
index 0000000000..d63b6f8a8d
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_traceid_ratio.py
@@ -0,0 +1,80 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from __future__ import annotations
+
+from typing import Sequence
+
+from opentelemetry.context import Context
+from opentelemetry.trace import Link, SpanKind, TraceState
+from opentelemetry.util.types import Attributes
+
+from ._composable import ComposableSampler, SamplingIntent
+from ._trace_state import serialize_th
+from ._util import INVALID_THRESHOLD, MAX_THRESHOLD, calculate_threshold
+
+
+class ComposableTraceIDRatioBased(ComposableSampler):
+    _threshold: int
+    _description: str
+
+    def __init__(self, ratio: float):
+        threshold = calculate_threshold(ratio)
+        if threshold == MAX_THRESHOLD:
+            threshold_str = "max"
+        else:
+            threshold_str = serialize_th(threshold)
+        if threshold != MAX_THRESHOLD:
+            intent = SamplingIntent(threshold=threshold)
+        else:
+            intent = SamplingIntent(
+                threshold=INVALID_THRESHOLD, threshold_reliable=False
+            )
+        self._intent = intent
+        self._description = f"ComposableTraceIDRatioBased{{threshold={threshold_str}, ratio={ratio}}}"
+
+    def sampling_intent(
+        self,
+        parent_ctx: Context | None,
+        name: str,
+        span_kind: SpanKind | None,
+        attributes: Attributes,
+        links: Sequence[Link] | None,
+        trace_state: TraceState | None,
+    ) -> SamplingIntent:
+        return self._intent
+
+    def get_description(self) -> str:
+        return self._description
+
+
+def composable_traceid_ratio_based(
+    ratio: float,
+) -> ComposableSampler:
+    """Returns a composable sampler that samples each span with a fixed ratio.
+
+    - Returns a SamplingIntent with threshold determined by the configured sampling ratio
+    - Sets threshold_reliable to true
+    - Does not add any attributes
+
+    Note:
+        If the ratio is 0, it will behave as an ComposableAlwaysOff sampler instead.
+
+    Args:
+        ratio: The sampling ratio to use (between 0.0 and 1.0).
+    """
+    if not 0.0 <= ratio <= 1.0:
+        raise ValueError("Sampling ratio must be between 0.0 and 1.0")
+
+    return ComposableTraceIDRatioBased(ratio)
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_util.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_util.py
new file mode 100644
index 0000000000..4e9fd7d234
--- /dev/null
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_util.py
@@ -0,0 +1,36 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+RANDOM_VALUE_BITS = 56
+MAX_THRESHOLD = 1 << RANDOM_VALUE_BITS  # 0% sampling
+MIN_THRESHOLD = 0  # 100% sampling
+MAX_RANDOM_VALUE = MAX_THRESHOLD - 1
+INVALID_THRESHOLD = -1
+INVALID_RANDOM_VALUE = -1
+
+_probability_threshold_scale = float.fromhex("0x1p56")
+
+
+def calculate_threshold(sampling_probability: float) -> int:
+    return MAX_THRESHOLD - round(
+        sampling_probability * _probability_threshold_scale
+    )
+
+
+def is_valid_threshold(threshold: int) -> bool:
+    return MIN_THRESHOLD <= threshold <= MAX_THRESHOLD
+
+
+def is_valid_random_value(random_value: int) -> bool:
+    return 0 <= random_value <= MAX_RANDOM_VALUE
diff --git a/opentelemetry-sdk/tests/conftest.py b/opentelemetry-sdk/tests/conftest.py
index 92fd7a734d..b0b633982e 100644
--- a/opentelemetry-sdk/tests/conftest.py
+++ b/opentelemetry-sdk/tests/conftest.py
@@ -12,8 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import random
 from os import environ
 
+import pytest
+
 from opentelemetry.environment_variables import OTEL_PYTHON_CONTEXT
 
 
@@ -25,3 +28,9 @@ def pytest_sessionstart(session):
 def pytest_sessionfinish(session):
     # pylint: disable=unused-argument
     environ.pop(OTEL_PYTHON_CONTEXT)
+
+
+@pytest.fixture(autouse=True)
+def random_seed():
+    # We use random numbers a lot in sampling tests, make sure they are always the same.
+    random.seed(0)
diff --git a/opentelemetry-sdk/tests/trace/composite_sampler/__init__.py b/opentelemetry-sdk/tests/trace/composite_sampler/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/opentelemetry-sdk/tests/trace/composite_sampler/test_always_off.py b/opentelemetry-sdk/tests/trace/composite_sampler/test_always_off.py
new file mode 100644
index 0000000000..0a03344f88
--- /dev/null
+++ b/opentelemetry-sdk/tests/trace/composite_sampler/test_always_off.py
@@ -0,0 +1,54 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from opentelemetry.sdk.trace._sampling_experimental import (
+    composable_always_off,
+    composite_sampler,
+)
+from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
+from opentelemetry.sdk.trace.sampling import Decision
+
+
+def test_description():
+    assert composable_always_off().get_description() == "ComposableAlwaysOff"
+
+
+def test_threshold():
+    assert (
+        composable_always_off()
+        .sampling_intent(None, "test", None, {}, None, None)
+        .threshold
+        == -1
+    )
+
+
+def test_sampling():
+    sampler = composite_sampler(composable_always_off())
+
+    num_sampled = 0
+    for _ in range(10000):
+        res = sampler.should_sample(
+            None,
+            RandomIdGenerator().generate_trace_id(),
+            "span",
+            None,
+            None,
+            None,
+            None,
+        )
+        if res.decision == Decision.RECORD_AND_SAMPLE:
+            num_sampled += 1
+        assert not res.trace_state
+
+    assert num_sampled == 0
diff --git a/opentelemetry-sdk/tests/trace/composite_sampler/test_always_on.py b/opentelemetry-sdk/tests/trace/composite_sampler/test_always_on.py
new file mode 100644
index 0000000000..a787b221a4
--- /dev/null
+++ b/opentelemetry-sdk/tests/trace/composite_sampler/test_always_on.py
@@ -0,0 +1,55 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from opentelemetry.sdk.trace._sampling_experimental import (
+    composable_always_on,
+    composite_sampler,
+)
+from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
+from opentelemetry.sdk.trace.sampling import Decision
+
+
+def test_description():
+    assert composable_always_on().get_description() == "ComposableAlwaysOn"
+
+
+def test_threshold():
+    assert (
+        composable_always_on()
+        .sampling_intent(None, "test", None, {}, None, None)
+        .threshold
+        == 0
+    )
+
+
+def test_sampling():
+    sampler = composite_sampler(composable_always_on())
+
+    num_sampled = 0
+    for _ in range(10000):
+        res = sampler.should_sample(
+            None,
+            RandomIdGenerator().generate_trace_id(),
+            "span",
+            None,
+            None,
+            None,
+            None,
+        )
+        if res.decision == Decision.RECORD_AND_SAMPLE:
+            num_sampled += 1
+            assert res.trace_state is not None
+            assert res.trace_state.get("ot", "") == "th:0"
+
+    assert num_sampled == 10000
diff --git a/opentelemetry-sdk/tests/trace/composite_sampler/test_sampler.py b/opentelemetry-sdk/tests/trace/composite_sampler/test_sampler.py
new file mode 100644
index 0000000000..4bd45fd45c
--- /dev/null
+++ b/opentelemetry-sdk/tests/trace/composite_sampler/test_sampler.py
@@ -0,0 +1,202 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import pytest
+from pytest import param as p
+
+from opentelemetry.sdk.trace._sampling_experimental import (
+    ComposableSampler,
+    composable_always_off,
+    composable_always_on,
+    composable_parent_threshold,
+    composable_traceid_ratio_based,
+    composite_sampler,
+)
+from opentelemetry.sdk.trace._sampling_experimental._trace_state import (
+    OtelTraceState,
+)
+from opentelemetry.sdk.trace._sampling_experimental._util import (
+    INVALID_RANDOM_VALUE,
+    INVALID_THRESHOLD,
+)
+from opentelemetry.sdk.trace.sampling import Decision
+from opentelemetry.trace import (
+    NonRecordingSpan,
+    SpanContext,
+    TraceFlags,
+    TraceState,
+    set_span_in_context,
+)
+
+TRACE_ID = int("00112233445566778800000000000000", 16)
+SPAN_ID = int("0123456789abcdef", 16)
+
+
+@dataclass
+class Input:
+    sampler: ComposableSampler
+    sampled: bool
+    threshold: int | None
+    random_value: int | None
+
+
+@dataclass
+class Output:
+    sampled: bool
+    threshold: int
+    random_value: int
+
+
+@pytest.mark.parametrize(
+    "input,output",
+    (
+        p(
+            Input(
+                sampler=composable_always_on(),
+                sampled=True,
+                threshold=None,
+                random_value=None,
+            ),
+            Output(
+                sampled=True, threshold=0, random_value=INVALID_RANDOM_VALUE
+            ),
+            id="min threshold no parent random value",
+        ),
+        p(
+            Input(
+                sampler=composable_always_on(),
+                sampled=True,
+                threshold=None,
+                random_value=0x7F99AA40C02744,
+            ),
+            Output(sampled=True, threshold=0, random_value=0x7F99AA40C02744),
+            id="min threshold with parent random value",
+        ),
+        p(
+            Input(
+                sampler=composable_always_off(),
+                sampled=True,
+                threshold=None,
+                random_value=None,
+            ),
+            Output(
+                sampled=False,
+                threshold=INVALID_THRESHOLD,
+                random_value=INVALID_RANDOM_VALUE,
+            ),
+            id="max threshold",
+        ),
+        p(
+            Input(
+                sampler=composable_parent_threshold(composable_always_on()),
+                sampled=False,
+                threshold=0x7F99AA40C02744,
+                random_value=0x7F99AA40C02744,
+            ),
+            Output(
+                sampled=True,
+                threshold=0x7F99AA40C02744,
+                random_value=0x7F99AA40C02744,
+            ),
+            id="parent based in consistent mode",
+        ),
+        p(
+            Input(
+                sampler=composable_parent_threshold(composable_always_on()),
+                sampled=True,
+                threshold=None,
+                random_value=None,
+            ),
+            Output(
+                sampled=True,
+                threshold=INVALID_THRESHOLD,
+                random_value=INVALID_RANDOM_VALUE,
+            ),
+            id="parent based in legacy mode",
+        ),
+        p(
+            Input(
+                sampler=composable_traceid_ratio_based(0.5),
+                sampled=True,
+                threshold=None,
+                random_value=0x7FFFFFFFFFFFFF,
+            ),
+            Output(
+                sampled=False,
+                threshold=INVALID_THRESHOLD,
+                random_value=0x7FFFFFFFFFFFFF,
+            ),
+            id="half threshold not sampled",
+        ),
+        p(
+            Input(
+                sampler=composable_traceid_ratio_based(0.5),
+                sampled=False,
+                threshold=None,
+                random_value=0x80000000000000,
+            ),
+            Output(
+                sampled=True,
+                threshold=0x80000000000000,
+                random_value=0x80000000000000,
+            ),
+            id="half threshold sampled",
+        ),
+        p(
+            Input(
+                sampler=composable_traceid_ratio_based(1.0),
+                sampled=False,
+                threshold=0x80000000000000,
+                random_value=0x80000000000000,
+            ),
+            Output(sampled=True, threshold=0, random_value=0x80000000000000),
+            id="parent violating invariant",
+        ),
+    ),
+)
+def test_sample(input: Input, output: Output):
+    parent_state = OtelTraceState.invalid()
+    if input.threshold is not None:
+        parent_state.threshold = input.threshold
+    if input.random_value is not None:
+        parent_state.random_value = input.random_value
+    parent_state_str = parent_state.serialize()
+    parent_trace_state = (
+        TraceState((("ot", parent_state_str),)) if parent_state_str else None
+    )
+    flags = (
+        TraceFlags(TraceFlags.SAMPLED)
+        if input.sampled
+        else TraceFlags.get_default()
+    )
+    parent_span_context = SpanContext(
+        TRACE_ID, SPAN_ID, False, flags, parent_trace_state
+    )
+    parent_span = NonRecordingSpan(parent_span_context)
+    parent_context = set_span_in_context(parent_span)
+
+    result = composite_sampler(input.sampler).should_sample(
+        parent_context, TRACE_ID, "name", trace_state=parent_trace_state
+    )
+
+    decision = Decision.RECORD_AND_SAMPLE if output.sampled else Decision.DROP
+    state = OtelTraceState.parse(result.trace_state)
+
+    assert result.decision == decision
+    assert state.threshold == output.threshold
+    assert state.random_value == output.random_value
diff --git a/opentelemetry-sdk/tests/trace/composite_sampler/test_traceid_ratio.py b/opentelemetry-sdk/tests/trace/composite_sampler/test_traceid_ratio.py
new file mode 100644
index 0000000000..ad0dc1f491
--- /dev/null
+++ b/opentelemetry-sdk/tests/trace/composite_sampler/test_traceid_ratio.py
@@ -0,0 +1,81 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+import pytest
+
+from opentelemetry.sdk.trace._sampling_experimental import (
+    composable_traceid_ratio_based,
+    composite_sampler,
+)
+from opentelemetry.sdk.trace._sampling_experimental._trace_state import (
+    OtelTraceState,
+)
+from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
+from opentelemetry.sdk.trace.sampling import Decision
+
+
+@pytest.mark.parametrize(
+    ("ratio", "threshold"),
+    (
+        (1.0, "0"),
+        (0.5, "8"),
+        (0.25, "c"),
+        (1e-300, "max"),
+        (0, "max"),
+    ),
+)
+def test_description(ratio: float, threshold: str):
+    assert (
+        composable_traceid_ratio_based(ratio).get_description()
+        == f"ComposableTraceIDRatioBased{{threshold={threshold}, ratio={ratio}}}"
+    )
+
+
+@pytest.mark.parametrize(
+    ("ratio", "threshold"),
+    (
+        (1.0, 0),
+        (0.5, 36028797018963968),
+        (0.25, 54043195528445952),
+        (0.125, 63050394783186944),
+        (0.0, 72057594037927936),
+        (0.45, 39631676720860364),
+        (0.2, 57646075230342348),
+        (0.13, 62690106812997304),
+        (0.05, 68454714336031539),
+    ),
+)
+def test_sampling(ratio: float, threshold: int):
+    sampler = composite_sampler(composable_traceid_ratio_based(ratio))
+
+    num_sampled = 0
+    for _ in range(10000):
+        res = sampler.should_sample(
+            None,
+            RandomIdGenerator().generate_trace_id(),
+            "span",
+            None,
+            None,
+            None,
+            None,
+        )
+        if res.decision == Decision.RECORD_AND_SAMPLE:
+            num_sampled += 1
+            assert res.trace_state is not None
+            otts = OtelTraceState.parse(res.trace_state)
+            assert otts.threshold == threshold
+            assert otts.random_value == -1
+
+    expected_num_sampled = int(10000 * ratio)
+    assert abs(num_sampled - expected_num_sampled) < 50
diff --git a/opentelemetry-sdk/tests/trace/composite_sampler/test_tracestate.py b/opentelemetry-sdk/tests/trace/composite_sampler/test_tracestate.py
new file mode 100644
index 0000000000..0af527cf3a
--- /dev/null
+++ b/opentelemetry-sdk/tests/trace/composite_sampler/test_tracestate.py
@@ -0,0 +1,69 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed 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.
+
+import pytest
+
+from opentelemetry.sdk.trace._sampling_experimental._trace_state import (
+    OtelTraceState,
+)
+from opentelemetry.trace import TraceState
+
+
+@pytest.mark.parametrize(
+    "input_str,output_str",
+    (
+        ("a", "a"),
+        ("#", "#"),
+        ("rv:1234567890abcd", "rv:1234567890abcd"),
+        ("rv:01020304050607", "rv:01020304050607"),
+        ("rv:1234567890abcde", ""),
+        ("th:1234567890abcd", "th:1234567890abcd"),
+        ("th:1234567890abcd", "th:1234567890abcd"),
+        ("th:10000000000000", "th:1"),
+        ("th:1234500000000", "th:12345"),
+        ("th:0", "th:0"),
+        ("th:100000000000000", ""),
+        ("th:1234567890abcde", ""),
+        pytest.param(
+            f"a:{'X' * 214};rv:1234567890abcd;th:1234567890abcd;x:3",
+            f"th:1234567890abcd;rv:1234567890abcd;a:{'X' * 214};x:3",
+            id="long",
+        ),
+        ("th:x", ""),
+        ("th:100000000000000", ""),
+        ("th:10000000000000", "th:1"),
+        ("th:1000000000000", "th:1"),
+        ("th:100000000000", "th:1"),
+        ("th:10000000000", "th:1"),
+        ("th:1000000000", "th:1"),
+        ("th:100000000", "th:1"),
+        ("th:10000000", "th:1"),
+        ("th:1000000", "th:1"),
+        ("th:100000", "th:1"),
+        ("th:10000", "th:1"),
+        ("th:1000", "th:1"),
+        ("th:100", "th:1"),
+        ("th:10", "th:1"),
+        ("th:1", "th:1"),
+        ("th:10000000000001", "th:10000000000001"),
+        ("th:10000000000010", "th:1000000000001"),
+        ("rv:x", ""),
+        ("rv:100000000000000", ""),
+        ("rv:10000000000000", "rv:10000000000000"),
+        ("rv:1000000000000", ""),
+    ),
+)
+def test_marshal(input_str: str, output_str: str):
+    state = OtelTraceState.parse(TraceState((("ot", input_str),))).serialize()
+    assert state == output_str