class_docstr: InputLoader that reads data in as a Spark DataFrame.
class_docstr_tok: ["InputLoader","that","reads","data","in","as","a","Spark","DataFrame","."]
docstr: Get s3 partitions filter condition based on input offsets.
  
  Args:
      input_start_offset (str): Start offset that is used to calculate the input start date.
      input_end_offset (str): End offset that is used to calculate the input end date.
  
  Returns:
      str: A SQL string that defines the condition of time range filter.
docstr_tok: ["Get","s3","partitions","filter","condition","based","on","input","offsets",".","Args",":","input_start_offset","(","str",")",":","Start","offset","that","is","used","to","calculate","the","input","start","date",".","input_end_offset","(","str",")",":","End","offset","that","is","used","to","calculate","the","input","end","date",".","Returns",":","str",":","A","SQL","string","that","defines","the","condition","of","time","range","filter","."]
code: def _get_s3_partitions_offset_filter_condition(
      self, input_start_offset: str, input_end_offset: str
  ) -> str:
      """Get s3 partitions filter condition based on input offsets.
  
      Args:
          input_start_offset (str): Start offset that is used to calculate the input start date.
          input_end_offset (str): End offset that is used to calculate the input end date.
  
      Returns:
          str: A SQL string that defines the condition of time range filter.
      """
      if input_start_offset is None and input_end_offset is None:
          return None
  
      offset_parser = InputOffsetParser(
          self.environment_helper.get_job_scheduled_time()
      )
      (
          start_year,
          start_month,
          start_day,
          start_hour,
      ) = offset_parser.get_offset_date_year_month_day_hour(
          input_start_offset
      )
      (
          end_year,
          end_month,
          end_day,
          end_hour,
      ) = offset_parser.get_offset_date_year_month_day_hour(
          input_end_offset
      )
  
      # Include all records that event time is between start_year and end_year
      start_year_include_condition = (
          f"year >= '{start_year}'" if input_start_offset else None
      )
      end_year_include_condition = (
          f"year <= '{end_year}'" if input_end_offset else None
      )
      year_include_condition = " AND ".join(
          filter(
              None,
              [
                  start_year_include_condition,
                  end_year_include_condition,
              ],
          )
      )
  
      # Exclude all records that the event time is earlier than the start or later than the end
      start_offset_exclude_condition = (
          f"(year = '{start_year}' AND month < '{start_month}') "
          f"OR (year = '{start_year}' AND month = '{start_month}' AND day < '{start_day}') "
          f"OR (year = '{start_year}' AND month = '{start_month}' AND day = '{start_day}' "
          f"AND hour < '{start_hour}')"
          if input_start_offset
          else None
      )
      end_offset_exclude_condition = (
          f"(year = '{end_year}' AND month > '{end_month}') "
          f"OR (year = '{end_year}' AND month = '{end_month}' AND day > '{end_day}') "
          f"OR (year = '{end_year}' AND month = '{end_month}' AND day = '{end_day}' "
          f"AND hour >= '{end_hour}')"
          if input_end_offset
          else None
      )
      offset_exclude_condition = " OR ".join(
          filter(
              None,
              [
                  start_offset_exclude_condition,
                  end_offset_exclude_condition,
              ],
          )
      )
  
      filter_condition = f"({year_include_condition}) AND NOT ({offset_exclude_condition})"
  
      logger.info(
          "The filter condition for hive feature group is %s.",
          filter_condition,
      )
  
      return filter_condition
  
imports: from __future__ import absolute_import
  import logging
  import re
  from abc import ABC, abstractmethod
  from typing import Generic, Optional, TypeVar, Union
  import attr
  from pyspark.sql import DataFrame
  from sagemaker import Session
  from sagemaker.feature_store.feature_processor._constants import FEATURE_GROUP_ARN_REGEX_PATTERN
  from sagemaker.feature_store.feature_processor._data_source import CSVDataSource, FeatureGroupDataSource, ParquetDataSource, IcebergTableDataSource
  from sagemaker.feature_store.feature_processor._spark_factory import SparkSessionFactory
  from sagemaker.feature_store.feature_processor._input_offset_parser import InputOffsetParser
  from sagemaker.feature_store.feature_processor._env import EnvironmentHelper
label_desc: Use image node_id 6 for calling the SparkDataFrameInputLoader obj's underlying member method code with example usage: obj._get_s3_partitions_offset_filter_condition(input_start_offset, input_end_offset) and returns: filter_condition, None