//---------------------------------------------------------------------------
//	Greenplum Database
//	Copyright (C) 2012 EMC Corp.
//
//	@filename:
//		CParseHandlerWindowSpec.cpp
//
//	@doc:
//		Implementation of the SAX parse handler class for parsing the
//		window specification node
//---------------------------------------------------------------------------

#include "naucrates/dxl/parser/CParseHandlerWindowSpec.h"

#include "naucrates/dxl/operators/CDXLOperatorFactory.h"
#include "naucrates/dxl/parser/CParseHandlerFactory.h"
#include "naucrates/dxl/parser/CParseHandlerScalarOp.h"
#include "naucrates/dxl/parser/CParseHandlerSortColList.h"
#include "naucrates/dxl/parser/CParseHandlerWindowFrame.h"
using namespace gpdxl;

XERCES_CPP_NAMESPACE_USE

//---------------------------------------------------------------------------
//	@function:
//		CParseHandlerWindowSpec::CParseHandlerWindowSpec
//
//	@doc:
//		Ctor
//
//---------------------------------------------------------------------------
CParseHandlerWindowSpec::CParseHandlerWindowSpec(
	CMemoryPool *mp, CParseHandlerManager *parse_handler_mgr,
	CParseHandlerBase *parse_handler_root)
	: CParseHandlerBase(mp, parse_handler_mgr, parse_handler_root),
	  m_part_by_colid_array(nullptr),
	  m_dxl_window_spec_gen(nullptr),
	  m_mdname(nullptr),
	  m_has_window_frame(false)
{
}

//---------------------------------------------------------------------------
//	@function:
//		CParseHandlerWindowSpec::StartElement
//
//	@doc:
//		Invoked by Xerces to process an opening tag
//
//---------------------------------------------------------------------------
void
CParseHandlerWindowSpec::StartElement(const XMLCh *const element_uri,
									  const XMLCh *const element_local_name,
									  const XMLCh *const element_qname,
									  const Attributes &attrs)
{
	if (0 ==
		XMLString::compareString(CDXLTokens::XmlstrToken(EdxltokenWindowSpec),
								 element_local_name))
	{
		GPOS_ASSERT(0 == this->Length());
		GPOS_ASSERT(nullptr == m_dxl_window_spec_gen);
		GPOS_ASSERT(nullptr == m_mdname);

		// parse alias from attributes
		const XMLCh *xml_alias =
			attrs.getValue(CDXLTokens::XmlstrToken(EdxltokenAlias));
		if (nullptr != xml_alias)
		{
			CWStringDynamic *str_alias =
				CDXLUtils::CreateDynamicStringFromXMLChArray(
					m_parse_handler_mgr->GetDXLMemoryManager(), xml_alias);
			m_mdname = GPOS_NEW(m_mp) CMDName(m_mp, str_alias);
			GPOS_DELETE(str_alias);
		}

		const XMLCh *xml_part_cols = CDXLOperatorFactory::ExtractAttrValue(
			attrs, EdxltokenPartKeys, EdxltokenPhysicalWindow);
		m_part_by_colid_array = CDXLOperatorFactory::ExtractIntsToUlongArray(
			m_parse_handler_mgr->GetDXLMemoryManager(), xml_part_cols,
			EdxltokenPartKeys, EdxltokenPhysicalWindow);
		GPOS_ASSERT(nullptr != m_part_by_colid_array);
	}
	else if (0 == XMLString::compareString(
					  CDXLTokens::XmlstrToken(EdxltokenScalarSortColList),
					  element_local_name))
	{
		// parse handler for the sorting column list
		CParseHandlerBase *sort_col_list_parse_handler =
			CParseHandlerFactory::GetParseHandler(
				m_mp, CDXLTokens::XmlstrToken(EdxltokenScalarSortColList),
				m_parse_handler_mgr, this);
		m_parse_handler_mgr->ActivateParseHandler(sort_col_list_parse_handler);

		// store parse handler
		this->Append(sort_col_list_parse_handler);
		sort_col_list_parse_handler->startElement(
			element_uri, element_local_name, element_qname, attrs);
	}
	else if (0 == XMLString::compareString(
					  CDXLTokens::XmlstrToken(EdxltokenWindowFrame),
					  element_local_name))
	{
		m_has_window_frame = true;

		// parse handler for the leading and trailing scalar values
		CParseHandlerBase *window_frame_parse_handler =
			CParseHandlerFactory::GetParseHandler(
				m_mp, CDXLTokens::XmlstrToken(EdxltokenWindowFrame),
				m_parse_handler_mgr, this);
		m_parse_handler_mgr->ActivateParseHandler(window_frame_parse_handler);

		// store parse handler
		this->Append(window_frame_parse_handler);
		window_frame_parse_handler->startElement(
			element_uri, element_local_name, element_qname, attrs);
	}
	else
	{
		CWStringDynamic *str = CDXLUtils::CreateDynamicStringFromXMLChArray(
			m_parse_handler_mgr->GetDXLMemoryManager(), element_local_name);
		GPOS_RAISE(gpdxl::ExmaDXL, gpdxl::ExmiDXLUnexpectedTag,
				   str->GetBuffer());
	}
}

//---------------------------------------------------------------------------
//	@function:
//		CParseHandlerWindowSpec::EndElement
//
//	@doc:
//		Invoked by Xerces to process a closing tag
//
//---------------------------------------------------------------------------
void
CParseHandlerWindowSpec::EndElement(const XMLCh *const,	 // element_uri,
									const XMLCh *const element_local_name,
									const XMLCh *const	// element_qname
)
{
	if (0 !=
		XMLString::compareString(CDXLTokens::XmlstrToken(EdxltokenWindowSpec),
								 element_local_name))
	{
		CWStringDynamic *str = CDXLUtils::CreateDynamicStringFromXMLChArray(
			m_parse_handler_mgr->GetDXLMemoryManager(), element_local_name);
		GPOS_RAISE(gpdxl::ExmaDXL, gpdxl::ExmiDXLUnexpectedTag,
				   str->GetBuffer());
	}
	// sorting columns
	CDXLNode *sort_col_list_dxlnode = nullptr;

	// window frame associated with the window key
	CDXLWindowFrame *window_frame = nullptr;

	if (1 == this->Length())
	{
		if (m_has_window_frame)
		{
			// In GPDB 5 and before, window specification cannot have a window frame specification
			// without sort columns. This changed in GPDB 6/Postgres 8.4+ where a query
			// select b,c, count(c) over (partition by b) from (select * from foo) s;
			// adds a window frame that is unbounded.
			CParseHandlerWindowFrame *window_frame_parse_handler =
				dynamic_cast<CParseHandlerWindowFrame *>((*this)[0]);
			window_frame = window_frame_parse_handler->GetWindowFrame();
		}
		else
		{
			CParseHandlerSortColList *sort_col_list_parse_handler =
				dynamic_cast<CParseHandlerSortColList *>((*this)[0]);
			sort_col_list_dxlnode =
				sort_col_list_parse_handler->CreateDXLNode();
			sort_col_list_dxlnode->AddRef();
		}
	}
	else if (2 == this->Length())
	{
		CParseHandlerSortColList *sort_col_list_parse_handler =
			dynamic_cast<CParseHandlerSortColList *>((*this)[0]);
		sort_col_list_dxlnode = sort_col_list_parse_handler->CreateDXLNode();
		sort_col_list_dxlnode->AddRef();

		CParseHandlerWindowFrame *window_frame_parse_handler =
			dynamic_cast<CParseHandlerWindowFrame *>((*this)[1]);
		window_frame = window_frame_parse_handler->GetWindowFrame();
	}
	m_dxl_window_spec_gen =
		GPOS_NEW(m_mp) CDXLWindowSpec(m_mp, m_part_by_colid_array, m_mdname,
									  sort_col_list_dxlnode, window_frame);

	// deactivate handler
	m_parse_handler_mgr->DeactivateHandler();
}

// EOF
