diff --git a/cpp/src/arrow/dataset/dataset.cc b/cpp/src/arrow/dataset/dataset.cc
index 548f90612b3..e3b81977e2b 100644
--- a/cpp/src/arrow/dataset/dataset.cc
+++ b/cpp/src/arrow/dataset/dataset.cc
@@ -20,7 +20,9 @@
 #include <memory>
 #include <utility>
 
+#include "arrow/dataset/filter.h"
 #include "arrow/dataset/scanner.h"
+#include "arrow/util/iterator.h"
 #include "arrow/util/stl.h"
 
 namespace arrow {
@@ -46,11 +48,10 @@ Status SimpleDataFragment::Scan(std::shared_ptr<ScanContext> scan_context,
   return Status::OK();
 }
 
-Status Dataset::Make(const std::vector<std::shared_ptr<DataSource>>& sources,
-                     const std::shared_ptr<Schema>& schema,
-                     std::shared_ptr<Dataset>* out) {
+Status Dataset::Make(std::vector<std::shared_ptr<DataSource>> sources,
+                     std::shared_ptr<Schema> schema, std::shared_ptr<Dataset>* out) {
   // TODO: Ensure schema and sources align.
-  *out = std::make_shared<Dataset>(sources, schema);
+  *out = std::make_shared<Dataset>(std::move(sources), std::move(schema));
 
   return Status::OK();
 }
@@ -61,5 +62,44 @@ Status Dataset::NewScan(std::unique_ptr<ScannerBuilder>* out) {
   return Status::OK();
 }
 
+bool DataSource::AssumePartitionExpression(
+    const std::shared_ptr<ScanOptions>& scan_options,
+    std::shared_ptr<ScanOptions>* simplified_scan_options) const {
+  DCHECK_NE(simplified_scan_options, nullptr);
+  if (scan_options == nullptr) {
+    // null scan options; no selector to simplify
+    *simplified_scan_options = scan_options;
+    return true;
+  }
+
+  auto c = SelectorAssume(scan_options->selector, partition_expression_);
+  DCHECK_OK(c.status());
+  auto expr = std::move(c).ValueOrDie();
+
+  bool trivial = true;
+  if (expr->IsNull() || (expr->IsTrivialCondition(&trivial) && !trivial)) {
+    // selector is not satisfiable; yield no fragments
+    return false;
+  }
+
+  auto copy = std::make_shared<ScanOptions>(*scan_options);
+  copy->selector = ExpressionSelector(std::move(expr));
+  *simplified_scan_options = std::move(copy);
+  return true;
+}
+
+DataFragmentIterator DataSource::GetFragments(std::shared_ptr<ScanOptions> scan_options) {
+  std::shared_ptr<ScanOptions> simplified_scan_options;
+  if (!AssumePartitionExpression(scan_options, &simplified_scan_options)) {
+    return MakeEmptyIterator<std::shared_ptr<DataFragment>>();
+  }
+  return GetFragmentsImpl(std::move(simplified_scan_options));
+}
+
+DataFragmentIterator SimpleDataSource::GetFragmentsImpl(
+    std::shared_ptr<ScanOptions> scan_options) {
+  return MakeVectorIterator(fragments_);
+}
+
 }  // namespace dataset
 }  // namespace arrow
diff --git a/cpp/src/arrow/dataset/dataset.h b/cpp/src/arrow/dataset/dataset.h
index 4e75112ad4a..b6f7ca7106a 100644
--- a/cpp/src/arrow/dataset/dataset.h
+++ b/cpp/src/arrow/dataset/dataset.h
@@ -24,7 +24,7 @@
 
 #include "arrow/dataset/type_fwd.h"
 #include "arrow/dataset/visibility.h"
-#include "arrow/util/iterator.h"
+#include "arrow/util/macros.h"
 
 namespace arrow {
 namespace dataset {
@@ -77,11 +77,32 @@ class ARROW_DS_EXPORT DataSource {
  public:
   /// \brief GetFragments returns an iterator of DataFragments. The ScanOptions
   /// controls filtering and schema inference.
-  virtual DataFragmentIterator GetFragments(std::shared_ptr<ScanOptions> options) = 0;
+  DataFragmentIterator GetFragments(std::shared_ptr<ScanOptions> options);
+
+  /// \brief An expression which evaluates to true for all data viewed by this DataSource.
+  /// May be null, which indicates no information is available.
+  const std::shared_ptr<Expression>& partition_expression() const {
+    return partition_expression_;
+  }
 
   virtual std::string type() const = 0;
 
   virtual ~DataSource() = default;
+
+ protected:
+  DataSource() = default;
+  explicit DataSource(std::shared_ptr<Expression> c)
+      : partition_expression_(std::move(c)) {}
+
+  virtual DataFragmentIterator GetFragmentsImpl(std::shared_ptr<ScanOptions> options) = 0;
+
+  /// Mutates a ScanOptions by assuming partition_expression_ holds for all yielded
+  /// fragments. Returns false if the selector is not satisfiable in this DataSource.
+  virtual bool AssumePartitionExpression(
+      const std::shared_ptr<ScanOptions>& scan_options,
+      std::shared_ptr<ScanOptions>* simplified_scan_options) const;
+
+  std::shared_ptr<Expression> partition_expression_;
 };
 
 /// \brief A DataSource consisting of a flat sequence of DataFragments
@@ -90,9 +111,7 @@ class ARROW_DS_EXPORT SimpleDataSource : public DataSource {
   explicit SimpleDataSource(DataFragmentVector fragments)
       : fragments_(std::move(fragments)) {}
 
-  DataFragmentIterator GetFragments(std::shared_ptr<ScanOptions> options) override {
-    return MakeVectorIterator(fragments_);
-  }
+  DataFragmentIterator GetFragmentsImpl(std::shared_ptr<ScanOptions> options) override;
 
   std::string type() const override { return "simple_data_source"; }
 
@@ -107,13 +126,12 @@ class ARROW_DS_EXPORT Dataset : public std::enable_shared_from_this<Dataset> {
   /// WARNING, this constructor is not recommend, use Dataset::Make instead.
   /// \param[in] sources one or more input data sources
   /// \param[in] schema a known schema to conform to, may be nullptr
-  explicit Dataset(const std::vector<std::shared_ptr<DataSource>>& sources,
-                   const std::shared_ptr<Schema>& schema)
-      : schema_(schema), sources_(sources) {}
+  explicit Dataset(std::vector<std::shared_ptr<DataSource>> sources,
+                   std::shared_ptr<Schema> schema)
+      : schema_(std::move(schema)), sources_(std::move(sources)) {}
 
-  static Status Make(const std::vector<std::shared_ptr<DataSource>>& sources,
-                     const std::shared_ptr<Schema>& schema,
-                     std::shared_ptr<Dataset>* out);
+  static Status Make(std::vector<std::shared_ptr<DataSource>> sourcs,
+                     std::shared_ptr<Schema> schema, std::shared_ptr<Dataset>* out);
 
   /// \brief Begin to build a new Scan operation against this Dataset
   Status NewScan(std::unique_ptr<ScannerBuilder>* out);
diff --git a/cpp/src/arrow/dataset/file_base.cc b/cpp/src/arrow/dataset/file_base.cc
index 20f4944c830..a335a2a679d 100644
--- a/cpp/src/arrow/dataset/file_base.cc
+++ b/cpp/src/arrow/dataset/file_base.cc
@@ -20,9 +20,11 @@
 #include <algorithm>
 #include <vector>
 
+#include "arrow/dataset/filter.h"
 #include "arrow/filesystem/filesystem.h"
 #include "arrow/io/interfaces.h"
 #include "arrow/io/memory.h"
+#include "arrow/util/iterator.h"
 #include "arrow/util/stl.h"
 
 namespace arrow {
@@ -47,18 +49,18 @@ Status FileBasedDataFragment::Scan(std::shared_ptr<ScanContext> scan_context,
 
 FileSystemBasedDataSource::FileSystemBasedDataSource(
     fs::FileSystem* filesystem, const fs::Selector& selector,
-    std::shared_ptr<FileFormat> format, std::shared_ptr<ScanOptions> scan_options,
+    std::shared_ptr<FileFormat> format, std::shared_ptr<Expression> partition_expression,
     std::vector<fs::FileStats> stats)
-    : filesystem_(filesystem),
+    : DataSource(std::move(partition_expression)),
+      filesystem_(filesystem),
       selector_(std::move(selector)),
       format_(std::move(format)),
-      scan_options_(std::move(scan_options)),
       stats_(std::move(stats)) {}
 
 Status FileSystemBasedDataSource::Make(fs::FileSystem* filesystem,
                                        const fs::Selector& selector,
                                        std::shared_ptr<FileFormat> format,
-                                       std::shared_ptr<ScanOptions> scan_options,
+                                       std::shared_ptr<Expression> partition_expression,
                                        std::unique_ptr<FileSystemBasedDataSource>* out) {
   std::vector<fs::FileStats> stats;
   RETURN_NOT_OK(filesystem->GetTargetStats(selector, &stats));
@@ -71,12 +73,25 @@ Status FileSystemBasedDataSource::Make(fs::FileSystem* filesystem,
   stats.resize(new_end - stats.begin());
 
   out->reset(new FileSystemBasedDataSource(filesystem, selector, std::move(format),
-                                           std::move(scan_options), std::move(stats)));
+                                           std::move(partition_expression),
+                                           std::move(stats)));
   return Status::OK();
 }
 
-DataFragmentIterator FileSystemBasedDataSource::GetFragments(
-    std::shared_ptr<ScanOptions> options) {
+Status FileSystemBasedDataSource::Make(fs::FileSystem* filesystem,
+                                       const fs::Selector& selector,
+                                       std::shared_ptr<FileFormat> format,
+                                       std::unique_ptr<FileSystemBasedDataSource>* out) {
+  return Make(filesystem, selector, std::move(format), nullptr, out);
+}
+
+DataFragmentIterator FileSystemBasedDataSource::GetFragmentsImpl(
+    std::shared_ptr<ScanOptions> scan_options) {
+  std::shared_ptr<ScanOptions> simplified_scan_options;
+  if (!AssumePartitionExpression(scan_options, &simplified_scan_options)) {
+    return MakeEmptyIterator<std::shared_ptr<DataFragment>>();
+  }
+
   struct Impl : DataFragmentIterator {
     Impl(fs::FileSystem* filesystem, std::shared_ptr<FileFormat> format,
          std::shared_ptr<ScanOptions> scan_options, std::vector<fs::FileStats> stats)
@@ -105,7 +120,7 @@ DataFragmentIterator FileSystemBasedDataSource::GetFragments(
     std::vector<fs::FileStats> stats_;
   };
 
-  return DataFragmentIterator(Impl(filesystem_, format_, options, stats_));
+  return DataFragmentIterator(Impl(filesystem_, format_, scan_options, stats_));
 }
 
 }  // namespace dataset
diff --git a/cpp/src/arrow/dataset/file_base.h b/cpp/src/arrow/dataset/file_base.h
index f3add99b409..440b36f75d9 100644
--- a/cpp/src/arrow/dataset/file_base.h
+++ b/cpp/src/arrow/dataset/file_base.h
@@ -170,23 +170,26 @@ class ARROW_DS_EXPORT FileSystemBasedDataSource : public DataSource {
  public:
   static Status Make(fs::FileSystem* filesystem, const fs::Selector& selector,
                      std::shared_ptr<FileFormat> format,
-                     std::shared_ptr<ScanOptions> scan_options,
                      std::unique_ptr<FileSystemBasedDataSource>* out);
 
-  std::string type() const override { return "directory"; }
+  static Status Make(fs::FileSystem* filesystem, const fs::Selector& selector,
+                     std::shared_ptr<FileFormat> format,
+                     std::shared_ptr<Expression> partition_expression,
+                     std::unique_ptr<FileSystemBasedDataSource>* out);
 
-  DataFragmentIterator GetFragments(std::shared_ptr<ScanOptions> options) override;
+  std::string type() const override { return "directory"; }
 
  protected:
+  DataFragmentIterator GetFragmentsImpl(std::shared_ptr<ScanOptions> options) override;
+
   FileSystemBasedDataSource(fs::FileSystem* filesystem, const fs::Selector& selector,
                             std::shared_ptr<FileFormat> format,
-                            std::shared_ptr<ScanOptions> scan_options,
+                            std::shared_ptr<Expression> partition_expression,
                             std::vector<fs::FileStats> stats);
 
   fs::FileSystem* filesystem_ = NULLPTR;
   fs::Selector selector_;
   std::shared_ptr<FileFormat> format_;
-  std::shared_ptr<ScanOptions> scan_options_;
   std::vector<fs::FileStats> stats_;
 };
 
diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index 77c3a71e2d3..e6298231b89 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -37,6 +37,7 @@ using RecordBatchReaderPtr = std::unique_ptr<RecordBatchReader>;
 // A set of RowGroup identifiers
 using RowGroupSet = std::vector<int>;
 
+// TODO(bkietz) refactor this to use ProjectedRecordBatchReader
 class ParquetScanTask : public ScanTask {
  public:
   static Status Make(RowGroupSet row_groups, const std::vector<int>& columns_projection,
@@ -128,7 +129,7 @@ class ParquetScanTaskIterator {
   }
 
   Status Next(ScanTaskPtr* task) {
-    auto partition = partitionner_.Next();
+    auto partition = partitioner_.Next();
 
     // Iteration is done.
     if (partition.size() == 0) {
@@ -145,7 +146,8 @@ class ParquetScanTaskIterator {
   static Status InferColumnProjection(const parquet::FileMetaData& metadata,
                                       const std::shared_ptr<ScanOptions>& options,
                                       std::vector<int>* out) {
-    // TODO(fsaintjacques): Compute intersection _and_ validity
+    // TODO(fsaintjacques): Compute intersection _and_ validity, could probably reuse
+    // RecordBatchProjector here
     *out = internal::Iota(metadata.num_columns());
 
     return Status::OK();
@@ -155,11 +157,11 @@ class ParquetScanTaskIterator {
                           std::shared_ptr<parquet::FileMetaData> metadata,
                           std::unique_ptr<parquet::arrow::FileReader> reader)
       : columns_projection_(columns_projection),
-        partitionner_(std::move(metadata)),
+        partitioner_(std::move(metadata)),
         reader_(std::move(reader)) {}
 
   std::vector<int> columns_projection_;
-  ParquetRowGroupPartitioner partitionner_;
+  ParquetRowGroupPartitioner partitioner_;
   std::shared_ptr<parquet::arrow::FileReader> reader_;
 };
 
diff --git a/cpp/src/arrow/dataset/file_parquet_test.cc b/cpp/src/arrow/dataset/file_parquet_test.cc
index 8bb333bb718..171d8cee1a7 100644
--- a/cpp/src/arrow/dataset/file_parquet_test.cc
+++ b/cpp/src/arrow/dataset/file_parquet_test.cc
@@ -185,5 +185,9 @@ TEST_F(TestParquetFileSystemBasedDataSource, Recursive) { this->Recursive(); }
 
 TEST_F(TestParquetFileSystemBasedDataSource, DeletedFile) { this->DeletedFile(); }
 
+TEST_F(TestParquetFileSystemBasedDataSource, PredicatePushDown) {
+  this->PredicatePushDown();
+}
+
 }  // namespace dataset
 }  // namespace arrow
diff --git a/cpp/src/arrow/dataset/file_test.cc b/cpp/src/arrow/dataset/file_test.cc
index b6063a4f295..19e938a5e8b 100644
--- a/cpp/src/arrow/dataset/file_test.cc
+++ b/cpp/src/arrow/dataset/file_test.cc
@@ -107,5 +107,9 @@ TEST_F(TestDummyFileSystemBasedDataSource, Recursive) { this->Recursive(); }
 
 TEST_F(TestDummyFileSystemBasedDataSource, DeletedFile) { this->DeletedFile(); }
 
+TEST_F(TestDummyFileSystemBasedDataSource, PredicatePushDown) {
+  this->PredicatePushDown();
+}
+
 }  // namespace dataset
 }  // namespace arrow
diff --git a/cpp/src/arrow/dataset/filter.cc b/cpp/src/arrow/dataset/filter.cc
index e8985dbeb7e..7e1ee82b22b 100644
--- a/cpp/src/arrow/dataset/filter.cc
+++ b/cpp/src/arrow/dataset/filter.cc
@@ -28,6 +28,7 @@
 #include "arrow/compute/context.h"
 #include "arrow/compute/kernels/boolean.h"
 #include "arrow/compute/kernels/compare.h"
+#include "arrow/dataset/dataset.h"
 #include "arrow/record_batch.h"
 #include "arrow/util/logging.h"
 #include "arrow/visitor_inline.h"
@@ -938,5 +939,33 @@ Result<std::shared_ptr<DataType>> FieldExpression::Validate(const Schema& schema
   return null();
 }
 
+Result<std::shared_ptr<Expression>> SelectorAssume(
+    const std::shared_ptr<DataSelector>& selector,
+    const std::shared_ptr<Expression>& given) {
+  if (selector == nullptr || selector->filters.size() == 0) {
+    return ScalarExpression::Make(true);
+  }
+
+  auto get_expression = [](const std::shared_ptr<Filter>& f) {
+    DCHECK_EQ(f->type(), FilterType::EXPRESSION);
+    return checked_cast<const ExpressionFilter&>(*f).expression();
+  };
+
+  auto out_expr = get_expression(selector->filters[0]);
+  for (size_t i = 1; i < selector->filters.size(); ++i) {
+    out_expr = and_(std::move(out_expr), get_expression(selector->filters[i]));
+  }
+
+  if (given == nullptr) {
+    return std::move(out_expr);
+  }
+  return out_expr->Assume(*given);
+}
+
+std::shared_ptr<DataSelector> ExpressionSelector(std::shared_ptr<Expression> e) {
+  return std::make_shared<DataSelector>(
+      DataSelector{FilterVector{std::make_shared<ExpressionFilter>(std::move(e))}});
+}
+
 }  // namespace dataset
 }  // namespace arrow
diff --git a/cpp/src/arrow/dataset/filter.h b/cpp/src/arrow/dataset/filter.h
index b305d4fd3d7..364fca0d2d4 100644
--- a/cpp/src/arrow/dataset/filter.h
+++ b/cpp/src/arrow/dataset/filter.h
@@ -21,6 +21,7 @@
 #include <string>
 #include <utility>
 
+#include "arrow/compute/kernel.h"
 #include "arrow/compute/kernels/compare.h"
 #include "arrow/dataset/type_fwd.h"
 #include "arrow/dataset/visibility.h"
@@ -402,5 +403,12 @@ inline FieldExpression operator"" _(const char* name, size_t name_length) {
 }
 }  // namespace string_literals
 
+ARROW_DS_EXPORT Result<std::shared_ptr<Expression>> SelectorAssume(
+    const std::shared_ptr<DataSelector>& selector,
+    const std::shared_ptr<Expression>& given);
+
+ARROW_DS_EXPORT std::shared_ptr<DataSelector> ExpressionSelector(
+    std::shared_ptr<Expression> e);
+
 }  // namespace dataset
 }  // namespace arrow
diff --git a/cpp/src/arrow/dataset/scanner.cc b/cpp/src/arrow/dataset/scanner.cc
index 1e5b8a8f2f7..a0f39d05e2b 100644
--- a/cpp/src/arrow/dataset/scanner.cc
+++ b/cpp/src/arrow/dataset/scanner.cc
@@ -20,6 +20,7 @@
 #include <algorithm>
 
 #include "arrow/dataset/dataset.h"
+#include "arrow/util/iterator.h"
 
 namespace arrow {
 namespace dataset {
diff --git a/cpp/src/arrow/dataset/scanner.h b/cpp/src/arrow/dataset/scanner.h
index 023290cc8b8..cfb661f4c0a 100644
--- a/cpp/src/arrow/dataset/scanner.h
+++ b/cpp/src/arrow/dataset/scanner.h
@@ -34,26 +34,27 @@ struct ARROW_DS_EXPORT ScanContext {
   MemoryPool* pool = arrow::default_memory_pool();
 };
 
-// TODO(wesm): API for handling of post-materialization filters. For
-// example, if the user requests [$col1 > 0, $col2 > 0] and $col1 is a
-// partition key, but $col2 is not, then the filter "$col2 > 0" must
-// be evaluated in-memory against the RecordBatch objects resulting
-// from the Scan
-
 class ARROW_DS_EXPORT ScanOptions {
  public:
-  virtual ~ScanOptions() = default;
+  ScanOptions() = default;
 
-  const std::shared_ptr<DataSelector>& selector() const { return selector_; }
+  ScanOptions(std::shared_ptr<DataSelector> selector, std::shared_ptr<Schema> schema,
+              std::vector<std::shared_ptr<FileScanOptions>> options = {})
+      : selector(std::move(selector)), schema(std::move(schema)) {}
 
-  const std::shared_ptr<Schema>& schema() const { return schema_; }
+  virtual ~ScanOptions() = default;
+
+  MemoryPool* pool() const { return pool_; }
 
- protected:
   // Filters
-  std::shared_ptr<DataSelector> selector_;
+  std::shared_ptr<DataSelector> selector;
 
   // Schema to which record batches will be reconciled
-  std::shared_ptr<Schema> schema_;
+  std::shared_ptr<Schema> schema;
+
+  MemoryPool* pool_ = default_memory_pool();
+
+  std::vector<std::shared_ptr<FileScanOptions>> options;
 };
 
 /// \brief Read record batches from a range of a single data fragment. A
diff --git a/cpp/src/arrow/dataset/test_util.h b/cpp/src/arrow/dataset/test_util.h
index 74b81b90641..459185e527f 100644
--- a/cpp/src/arrow/dataset/test_util.h
+++ b/cpp/src/arrow/dataset/test_util.h
@@ -15,17 +15,20 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <functional>
 #include <memory>
 #include <string>
 #include <utility>
 #include <vector>
 
 #include "arrow/dataset/file_base.h"
+#include "arrow/dataset/filter.h"
 #include "arrow/filesystem/localfs.h"
 #include "arrow/filesystem/path_util.h"
 #include "arrow/record_batch.h"
 #include "arrow/testing/gtest_util.h"
 #include "arrow/util/io_util.h"
+#include "arrow/util/iterator.h"
 #include "arrow/util/stl.h"
 
 namespace arrow {
@@ -162,7 +165,12 @@ class FileSystemBasedDataSourceMixin : public FileSourceFixtureMixin {
   virtual std::vector<std::string> file_names() const = 0;
 
   void SetUp() override {
+    selector_.base_dir = "/";
+    selector_.recursive = true;
+
     format_ = std::make_shared<Format>();
+    schema_ = schema({field("dummy", null())});
+    options_ = std::make_shared<ScanOptions>();
 
     ASSERT_OK(
         TemporaryDir::Make("test-fsdatasource-" + format_->name() + "-", &temp_dir_));
@@ -174,6 +182,8 @@ class FileSystemBasedDataSourceMixin : public FileSourceFixtureMixin {
     for (auto path : file_names()) {
       CreateFile(path, "");
     }
+
+    partition_expression_ = ScalarExpression::Make(true);
   }
 
   void CreateFile(std::string path, std::string contents) {
@@ -188,69 +198,72 @@ class FileSystemBasedDataSourceMixin : public FileSourceFixtureMixin {
 
   void MakeDataSource() {
     ASSERT_OK(FileSystemBasedDataSource::Make(fs_.get(), selector_, format_,
-                                              std::make_shared<ScanOptions>(), &source_));
+                                              partition_expression_, &source_));
   }
 
  protected:
+  std::function<Status(std::shared_ptr<DataFragment> fragment)> OpenFragments(
+      size_t* count) {
+    return [this, count](std::shared_ptr<DataFragment> fragment) {
+      auto file_fragment =
+          internal::checked_pointer_cast<FileBasedDataFragment>(fragment);
+      ++*count;
+      auto extension =
+          fs::internal::GetAbstractPathExtension(file_fragment->source().path());
+      EXPECT_TRUE(format_->IsKnownExtension(extension));
+      std::shared_ptr<io::RandomAccessFile> f;
+      return this->fs_->OpenInputFile(file_fragment->source().path(), &f);
+    };
+  }
+
   void NonRecursive() {
-    selector_.base_dir = "/";
+    selector_.recursive = false;
     MakeDataSource();
 
-    int count = 0;
-    ASSERT_OK(
-        source_->GetFragments({}).Visit([&](std::shared_ptr<DataFragment> fragment) {
-          auto file_fragment =
-              internal::checked_pointer_cast<FileBasedDataFragment>(fragment);
-          ++count;
-          auto extension =
-              fs::internal::GetAbstractPathExtension(file_fragment->source().path());
-          EXPECT_TRUE(format_->IsKnownExtension(extension));
-          std::shared_ptr<io::RandomAccessFile> f;
-          return this->fs_->OpenInputFile(file_fragment->source().path(), &f);
-        }));
-
+    size_t count = 0;
+    ASSERT_OK(source_->GetFragments(options_).Visit(OpenFragments(&count)));
     ASSERT_EQ(count, 1);
   }
 
   void Recursive() {
-    selector_.base_dir = "/";
-    selector_.recursive = true;
     MakeDataSource();
 
-    int count = 0;
-    ASSERT_OK(
-        source_->GetFragments({}).Visit([&](std::shared_ptr<DataFragment> fragment) {
-          auto file_fragment =
-              internal::checked_pointer_cast<FileBasedDataFragment>(fragment);
-          ++count;
-          auto extension =
-              fs::internal::GetAbstractPathExtension(file_fragment->source().path());
-          EXPECT_TRUE(format_->IsKnownExtension(extension));
-          std::shared_ptr<io::RandomAccessFile> f;
-          return this->fs_->OpenInputFile(file_fragment->source().path(), &f);
-        }));
-
-    ASSERT_EQ(count, 4);
+    size_t count = 0;
+    ASSERT_OK(source_->GetFragments(options_).Visit(OpenFragments(&count)));
+    ASSERT_EQ(count, file_names().size());
   }
 
   void DeletedFile() {
-    selector_.base_dir = "/";
-    selector_.recursive = true;
     MakeDataSource();
     ASSERT_GT(file_names().size(), 0);
     ASSERT_OK(this->fs_->DeleteFile(file_names()[0]));
 
-    ASSERT_RAISES(
-        IOError,
-        source_->GetFragments({}).Visit([&](std::shared_ptr<DataFragment> fragment) {
-          auto file_fragment =
-              internal::checked_pointer_cast<FileBasedDataFragment>(fragment);
-          auto extension =
-              fs::internal::GetAbstractPathExtension(file_fragment->source().path());
-          EXPECT_TRUE(format_->IsKnownExtension(extension));
-          std::shared_ptr<io::RandomAccessFile> f;
-          return this->fs_->OpenInputFile(file_fragment->source().path(), &f);
-        }));
+    size_t count = 0;
+    ASSERT_RAISES(IOError, source_->GetFragments(options_).Visit(OpenFragments(&count)));
+  }
+
+  void PredicatePushDown() {
+    partition_expression_ = equal(field_ref("alpha"), ScalarExpression::Make<int16_t>(3));
+    MakeDataSource();
+
+    options_->selector = std::make_shared<DataSelector>();
+    options_->selector->filters.resize(1);
+
+    // with a filter identical to the partition condition, all fragments are yielded
+    options_->selector->filters[0] =
+        std::make_shared<ExpressionFilter>(partition_expression_->Copy());
+
+    size_t count = 0;
+    // ASSERT_OK(source_->GetFragments(context_)->Visit(OpenFragments(&count)));
+    // ASSERT_EQ(count, file_names().size());
+
+    // with a filter which contradicts the partition condition, no fragments are yielded
+    options_->selector->filters[0] = std::make_shared<ExpressionFilter>(
+        equal(field_ref("alpha"), ScalarExpression::Make<int16_t>(0)));
+
+    count = 0;
+    ASSERT_OK(source_->GetFragments(options_).Visit(OpenFragments(&count)));
+    ASSERT_EQ(count, 0);
   }
 
   fs::Selector selector_;
@@ -259,6 +272,9 @@ class FileSystemBasedDataSourceMixin : public FileSourceFixtureMixin {
   std::shared_ptr<fs::FileSystem> fs_;
   std::unique_ptr<TemporaryDir> temp_dir_;
   std::shared_ptr<FileFormat> format_;
+  std::shared_ptr<Schema> schema_;
+  std::shared_ptr<ScanOptions> options_;
+  std::shared_ptr<Expression> partition_expression_;
 };
 
 template <typename Gen>
diff --git a/cpp/src/arrow/dataset/type_fwd.h b/cpp/src/arrow/dataset/type_fwd.h
index 4f195334e2f..db1ebb17130 100644
--- a/cpp/src/arrow/dataset/type_fwd.h
+++ b/cpp/src/arrow/dataset/type_fwd.h
@@ -53,6 +53,7 @@ class FileFormat;
 class FileScanOptions;
 class FileWriteOptions;
 
+class Expression;
 class Filter;
 using FilterVector = std::vector<std::shared_ptr<Filter>>;
 
diff --git a/cpp/src/arrow/util/iterator.h b/cpp/src/arrow/util/iterator.h
index 38ad382eb70..20f865cebf2 100644
--- a/cpp/src/arrow/util/iterator.h
+++ b/cpp/src/arrow/util/iterator.h
@@ -25,6 +25,7 @@
 #include "arrow/status.h"
 #include "arrow/util/functional.h"
 #include "arrow/util/macros.h"
+#include "arrow/util/visibility.h"
 
 namespace arrow {
 
@@ -45,9 +46,15 @@ class Iterator {
  public:
   /// \brief Iterator may be constructed from any type which has a member function
   /// with signature Status Next(T*);
+  ///
   /// The argument is moved or copied to the heap and kept in a unique_ptr<void>. Only
   /// its destructor and its Next method (which are stored in function pointers) are
   /// referenced after construction.
+  ///
+  /// This approach is used to dodge MSVC linkage hell (ARROW-6244, ARROW-6558) when using
+  /// an abstract template base class: instead of being inlined as usual for a template
+  /// function the base's virtual destructor will be exported, leading to multiple
+  /// definition errors when linking to any other TU where the base is instantiated.
   template <typename Wrapped>
   explicit Iterator(Wrapped has_next)
       : ptr_(new Wrapped(std::move(has_next)), Delete<Wrapped>), next_(Next<Wrapped>) {}
diff --git a/cpp/src/arrow/util/visibility.h b/cpp/src/arrow/util/visibility.h
index 95cd9cf5ba2..2b3751ed096 100644
--- a/cpp/src/arrow/util/visibility.h
+++ b/cpp/src/arrow/util/visibility.h
@@ -34,12 +34,14 @@
 #endif
 
 #define ARROW_NO_EXPORT
+#define ARROW_FORCE_INLINE __forceinline
 #else  // Not Windows
 #ifndef ARROW_EXPORT
 #define ARROW_EXPORT __attribute__((visibility("default")))
 #endif
 #ifndef ARROW_NO_EXPORT
 #define ARROW_NO_EXPORT __attribute__((visibility("hidden")))
+#define ARROW_FORCE_INLINE
 #endif
 #endif  // Non-Windows